阅读envoy_self.process.poll

作者：IT小白 | 2024-07-31 08:27:25

踩

self.process.poll

一、阅读背景

18年最后一个工作日，前一天便开启了划水模式，倒也没有浪费时间，读了envoy的源码。

envoy的作者是Kenneth Reitz，代码256行，envoy地址。

二、阅读备注


"""
envoy.core
~~~~~~~~~~
 
This module provides envoy awesomeness.
"""

作者开头留下这么句话，百度翻译过来时”这个模块提供使者的敬畏“。实际上，这个模块对subprocess进行了封装。

导入的模块


import os  # 负责与操作系统的交互
import sys  # 负责与python解释器的交互
import shlex  # 解析简单的类shell语言或解析加引号的字符串
import signal  # 用于处理信号，主要在unix系统上使用（信号----进程之间的通信方式，是一种软件中断。一个进程一旦接到某个信号就会打断原来的程序执行流程并先处理信号）
import subprocess  # 允许用户创建新的进程，并与它进行通信，获取标准的输入、输出、标准错误和返回码等
import threading  # 线程
import traceback  # 捕获并打印异常

包相关信息


__version__ = '0.0.3'  # 版本
__license__ = 'MIT'   # 许可证
__author__ = 'Kenneth Reitz'  # 作者

终止进程函数(一)


def _terminate_process(process):  # 终止进程的函数
    if sys.platform == 'win32':  # 判断平台是不是windows32位系统
        import ctypes  # python的一个外部库，可使用python语言调用已编译好的C语言函数以及数据类型进行数据交换
        PROCESS_TERMINATE = 1  # PROCESS_TERMINATE翻译为进程终止
        handle = ctypes.windll.kernel32.OpenProcess(PROCESS_TERMINATE, False, process.pid)  # 暂时看不懂，目测为杀掉进程
        ctypes.windll.kernel32.TerminateProcess(handle, -1)
        ctypes.windll.kernel32.CloseHandle(handle)
    else:
        os.kill(process.pid, signal.SIGTERM)  # 直接杀掉进程，只在unix上有用，win上用sys.exit()或os._exit()退出程序。      process.pid为进程的pid，signal.SIGTERM为signal的信号码

终止进程函数(二)


def _kill_process(process):  # 终止进程函数
    if sys.platform == 'win32':
        _terminate_process(process)  # 平台为windows32位系统才调用(为了独立函数的完备性，不是太能理解)
    else:
        os.kill(process.pid, signal.SIGKILL)

判断进程存活函数


# 判断进程是否存活的函数(这里是为了兼容性，不同版本实现的方法不一样)
def _is_alive(thread):  # 判断进程是否存活的函数(这里是为了兼容性，不同版本实现的方法不一样)
    if hasattr(thread, "is_alive"):  # 判断进程是否有is_alive属性，
        return thread.is_alive()
    else:
        return thread.isAlive()

Command类


class Command(object):
    def __init__(self, cmd):  # 初始化
        self.cmd = cmd  # 类属性
        self.process = None
        self.out = None
        self.err = None
        self.returncode = None
        self.data = None
        self.exc = None
 
    def run(self, data, timeout, kill_timeout, env, cwd): 
        self.data = data
        environ = dict(os.environ)  # os.environ获取系统环境变量
        environ.update(env or {})  # 如果传入了env就把它加入environ中
 
        def target():
 
            try:
                self.process = subprocess.Popen(self.cmd,  # subprocess.Popen调用系统命令
                    universal_newlines=True,  # true表示各种换行符统一处理为‘\n’
                    shell=False,  # 设置为true表示指定的命令会在shell中执行
                    env=environ,  # 设置子程序的环境变量
                    stdin=subprocess.PIPE,  # 子程序的标准输入。subprocess.PIPE专用于Popen的stdin，stdout，stderr，表示需要创建一个新的管道
                    stdout=subprocess.PIPE,  # 子程序的标准输出
                    stderr=subprocess.PIPE,  # 子程序的标准错误
                    bufsize=0,  # 如果指定了bufsize参数作用就和内建函数open()一样：0表示不缓冲，1表示行缓冲，其他正数表示近似的缓冲区字节数，负数表示使用系统默认值。默认0
                    cwd=cwd,  # 设置子程序的当前目录
                )
 
                if sys.version_info[0] >= 3:  # sys.version_info获取版本号，返回含有5个元素的元祖sys.version_info(major=3, minor=6, micro=5, releaselevel='final', serial=0)
                    self.out, self.err = self.process.communicate(  # Popen.communicate(input=None)和子进程交互，发送数据到stdin，并从stdout何stderr读数据，直到接收EOF
                        input=bytes(self.data, "UTF-8") if self.data else None
                    )  # 大于3的版本就是要传入的data用bytes编码成字节作为参数
                else:
                    self.out, self.err = self.process.communicate(self.data)  # 不是大于3的版本不用编码直接作为参数
            except Exception as exc:  # 如果出现异常，将exc赋值给selfexc
                self.exc = exc
 
 
        thread = threading.Thread(target=target)  # 传一个可调用函数给Thread，并实例化创建一个线程
        thread.start()  # 启动线程
 
        thread.join(timeout)  # join方法是线程等待超时时间，如果在timeout时间内，线程还未结束，则不再阻塞线程
        if self.exc:  # 如果存在则抛出异常
            raise self.exc
        if _is_alive(thread):  # 如果线程还存活，则调用_terminate_process函数终止线程
            _terminate_process(self.process)
            thread.join(kill_timeout)  # 再加入一个超时时间，如果超时就再终止线程一次，最后再加入一个超时时间
            if _is_alive(thread):
                _kill_process(self.process)
                thread.join()
        self.returncode = self.process.returncode  # Popen.returncode子程序的返回值，由poll()或者wait()设置，间接地也由communicate()设置。为None表示子进程还没终止，为-N的话，表示进程被N信号终止(仅限于unix)
        return self.out, self.err  # 返回标准输出和标准错误

ConnectedCommand类


class ConnectedCommand(object):
    def __init__(self,
        process=None,
        std_in=None,
        std_out=None,
        std_err=None):
 
        self._process = process
        self.std_in = std_in
        self.std_out = std_out
        self.std_err = std_out
        self._status_code = None
 
    def __enter__(self):  # 上下文管理协议中有__enter__和__exit__，with开始运行时出发__enter__
        return self
 
    def __exit__(self, type, value, traceback):  # with运行结束后出发__exit__
        self.kill()
 
    @property
    def status_code(self):  # 只读status_code，返回进程状态码，返回None则表示进程仍在运行
        """The status code of the process.
        If the code is None, assume that it's still running.
        """
        return self._status_code
 
    @property
    def pid(self):  # 只读pid，返回进程的pid
        """The process' PID."""
        return self._process.pid
 
    def kill(self):  # 杀死子进程
        """Kills the process."""
        return self._process.kill()
 
    def expect(self, bytes, stream=None):  # 获取标准输出
        """Block until given bytes appear in the stream."""
        if stream is None:
            stream = self.std_out
 
    def send(self, str, end='\n'):  # 向标准输入里写入一行内容
        """Sends a line to std_in."""
        return self._process.stdin.write(str+end)
 
    def block(self):  # 等待子进程结束，返回returncode属性赋值给self._status_code
        """Blocks until command finishes. Returns Response instance."""
        self._status_code = self._process.wait()

Response类


class Response(object):  # 命令的响应
    """A command's response"""
 
    def __init__(self, process=None):
        super(Response, self).__init__()
 
        self._process = process
        self.command = None
        self.std_err = None
        self.std_out = None
        self.status_code = None
        self.history = []
 
    def __repr__(self):  # Response类字符串表示形式
        if len(self.command):
            return '<Response [{0}]>'.format(self.command[0])
        else:
            return '<Response>'

解析命令函数


# 解析命令字符串 ，返回list
def expand_args(command):  # 解析命令字符串 ，返回list
    """Parses command strings and returns a Popen-ready list."""
 
    # Prepare arguments.
    if isinstance(command, str):  # 判断命令是字符串还是unicode，py3中str即uncode字符串，合并了unicode
        splitter = shlex.shlex(command.encode('utf-8'))
        splitter.whitespace = '|'  # 把|当做空格处理
        splitter.whitespace_split = True  # 以|为切分依据
        command = []
 
        while True:
            token = splitter.get_token()  # 把切分的结果存到token中
            if token:  # 如果token存在，就加入command列表中
                command.append(token)
            else:
                break
 
        command = list(map(shlex.split, command))  # 将shlex.split作用于command列表中每一个元素，并赋值给command
 
    return command  # 返回

执行命令函数


# 执行给定的命令，返回Response，。直到进程结束或超时才会解锁
def run(command, data=None, timeout=None, kill_timeout=None, env=None, cwd=None):
    """Executes a given commmand and returns Response.
 
    Blocks until process is complete, or timeout is reached.
    """  # 执行给定的命令，返回Response，。直到进程结束或超时才会解锁
 
    command = expand_args(command)  # 得到命令列表
 
    history = []
    for c in command:
 
        if len(history):
            # due to broken pipe problems pass only first 10 KiB
            data = history[-1].std_out[0:10*1024]
 
        cmd = Command(c)  # 实例化Command，并赋值给cmd
        try:
            out, err = cmd.run(data, timeout, kill_timeout, env, cwd)  # 调用Command类的run方法执行命令，得到输出和标准错误
            status_code = cmd.returncode  # 得到子程序的返回值
        except OSError as e:  # 运行可能会有异常
            out, err = '', u"\n".join([e.strerror, traceback.format_exc()])
            status_code = 127
 
        r = Response(process=cmd)  # 实例化Response
 
        r.command = c
        r.std_out = out
        r.std_err = err
        r.status_code = status_code
 
        history.append(r)  # 把r加入history列表
 
    r = history.pop()  # 从列表中取出最后一个元素
    r.history = history
 
    return r

建立进程函数


# 用给定命令建立一个进程
def connect(command, data=None, env=None, cwd=None):  # 用给定命令建立一个进程
    """Spawns a new process from the given command."""
 
    # TODO: support piped commands
    command_str = expand_args(command).pop()  # 调用expand_args解析输入命令，并将返回list中最后一个元素赋值给command_str
    environ = dict(os.environ)  # 字典化环境变量
    environ.update(env or {})  # 字典增加传入的环境变量
 
    process = subprocess.Popen(command_str,
        universal_newlines=True,
        shell=False,
        env=environ,
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        bufsize=0,
        cwd=cwd,
    )
 
    return ConnectedCommand(process=process)  # 返回ConnectedCommand实例

三、最后

写在最后，目前读源码还是有一定压力，250多行两个多小时读完还不能第一时间消化完。昨天看到一个方法”边阅读源码，边自己尝试实现“，也觉得比较科学，接下来便采用这种方式，慢慢读，慢慢做笔记。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/IT小白/article/detail/907864