首页 学习 编程语言 python

以下为个人学习笔记整理

# HotFix 热更新

# 概念：

  • 热更新是指在 Python 程序运行过程中，修改代码中的部分片段，并能够不需要重新启动程序，便能够在运行程序中生效。
  • 热更新一般都是基于 module 来进行的，所以热更新本质就是更新 module
  • 为了能够保证热更之前创建的绝大多数对象是能够正常工作的，一般会尽量避免对对象直接进行替换，能修改的尽量不替换。

# 介绍：

  • 一个 module 里面包含的内容大致可以分为以下几种：
    • class ：类
    • function ：方法
    • global object ：全局对象

看上去只有三种类似的对象，其实更新的时候注意的点还挺多的：

  • 模块是新增或者类型发生了变更，可以直接替换。
  • 一般不对模块内建函数和内建全局对象做操作（大部分是不可修改的，还有一些平时也不会改，没有热更必要）。
  • 除此以外的就是对三种类型的分别更新了。
def reload_module(module_name:str):
    """
    热更模块(module):
        ①: 更新新增成员
        ②: 跳过 builtins 模块
        ③: 处理类型不同的成员
        ④: 更新类成员
        ⑤: 更新函数成员
        ⑥: 更新成员变量
    """
	# 该模块之前没有被加载，不允许热更新
    old_module = sys.modules.get(module_name, None)
    if not old_module:
        raise Exception(f"{module_name} is not import can't reload")
    
    # Python3.7 的机制，如果不 pop 掉旧的模块， import_module 操作只会从 moduels 取出旧的缓存数据，不会重新构建
    sys.modules.pop(module_name)
    new_module = importlib.import_module(module_name)
    if not inspect.ismodule(new_module):
        raise Exception(f"{new_module.__name__} is not a module")
    for name, new_member in inspect.getmembers(new_module):
        # 模块名称和 member 的 key 可能不一致 例如 import xxx as x
        member_name = getattr(new_member, "__name__", name)
        old_member = old_module.__dict__.get(name, None)
        # 模块内的成员模块 和 built-in 函数不做处理
        if inspect.ismodule(new_member) or \
            inspect.isbuiltin(new_member) or \
            member_name in builtins.__dict__:
            continue
        
        # 原模块没有的内容或者类型不同直接换
        elif not old_member or type(old_member) != type(new_member):
            setattr(old_module, name, new_member)
        # 类，走类自己的热更
        elif inspect.isclass(new_member):
            # 枚举类型强制替换，因为枚举定义后无法被修改，这里只能替换类的定义。
            if issubclass(type(old_member), enum.EnumMeta):
                setattr(old_module, name, new_member)
            else:
                reload_class(old_member, new_member)
        # 函数，热更之
        elif inspect.isfunction(new_member):
            # 热更失败直接换，失败的原因可能是函数本身的闭包参数变更:
            # 也不是说热更失败，只是这种情况下，热更也没办法兼容旧的逻辑，毫无意义，徒增烦恼
            # 	@装饰器				------>		@装饰器
            # 	def func(arg1, arg2)			  def func(arg1)
            if not reload_func(old_member, new_member):
                setattr(old_module, name, new_member)
        else:
            setattr(old_module, name, new_member)
    sys.modules[module_name] = old_module
下面就来介绍一下 class 、 function 、 global variable 的热更新问题。
# 热更 Class：
热更新类的一些注意事项：
  • 类的更新一般不进行替换，而是把新的类中的内容更新到旧的类里面，为了兼容一些已经创建的类实体。
  • 对于以前的旧类中存在而新类不存在的内容，根据自身需求选择是否保留（这里删掉了）。
  • 类中包含部分不能直接修改的对象，跳过它们的更新：
    • '__dict__', '__doc__', '__self__', '__func__' 这些都是不可修改的。
  • 和模块一样，把新增的内容和类型不一致的内容更新到旧的类里面。
  • 和模块一样，跳过内建函数。
  • staticmethod、classmethod 函数由于无法修改，只能单独热更 __func__ 字段。
  • property 修饰的函数直接替换即可。
  • 类函数则走函数的正常更新流程。
  • 类中定义的类依旧走类的更新。
  • 其他内容直接覆盖即可。
def reload_class(old_class:type, new_class:type):
    """
    热更类(class):
        ①: 新增成员直接加
        ②: builtins 成员不处理
        ③: methoddescriptor 成员不处理
        ④: 类型不同直接替换
        ⑤: staticmethod,classmethod,property,method 直接更新
        ⑥: function 直接更新
        ⑦: class 递归更新
        ⑧: class 属性成员直接更新
    """
    # 删除新类不存在的旧类成员
    for name, attr in list(old_class.__dict__.items()):
        if name in new_class.__dict__: 
            continue
        if not inspect.isfunction(attr): 
            continue
        type.__delattr__(old_class, name)
    ignore_attr_lst = [ "__dict__",         # attribute objects is not writable
        '__doc__', '__self__', '__func__',    # can't set attributes of built-in/extension
    ]
    
    for name, new_attr in new_class.__dict__.items():
        if name in ignore_attr_lst:
            continue
        old_attr = old_class.__dict__.get(name, None)
        # 新增内容直接加
        if not old_attr:
            setattr(old_class, name, new_attr)
        elif inspect.isbuiltin(new_attr) or name in builtins.__dict__.keys():
            continue
        # 类型不同直接换
        elif type(old_attr) != type(new_attr):
            setattr(old_class, name, new_attr)
        elif isinstance(new_attr, (staticmethod, classmethod)):
            if not reload_func(old_attr.__func__, new_attr.__func__):
                setattr(old_class, name, new_attr)
        elif isinstance(new_attr, property):
            setattr(old_class, name, new_attr)
        elif inspect.isfunction(new_attr):
            if not reload_func(old_attr, new_attr):
                setattr(old_class, name, new_attr)
        elif inspect.isclass(new_attr):
            reload_class(old_attr, new_attr)
        else:
            setattr(old_class, name, new_attr)
# 热更 Function：
函数的更新算是热更里面最核心的内容了，注意点也挺多：
  • 函数的热更新一般也不对函数本身进行替换，直接修改即可，迫不得已情况下可以考虑换掉。
  • 函数本身因为没有涉及到过多的自定义内容，大部分都是逻辑，所以内置的东西粗略的看下来就几样：
    • __closure__ ：闭包的关联参数
    • __code__ ：编译后的代码对象
    • __defaults__ ：k-v 的默认值
    • __dict__ ：命名空间支持的函数属性
    • __globals__ ：全局变量字典
    • __name__ ：函数名
    • __qualname__ ：函数全名
    • __annotations__ ：类型标注
    • __kwdefaults__ ：关键字默认值字典
  • 把上述的几个替换一下即可，这里要注意一下，有部分字段也是不可修改的：
    • __class__ ：assignment only supported for heap types or ModuleType subclasses
    • __closure__ ：readonly attribute
    • __globals__ ：readonly attribute
  • 除此以外，还需要注意检验函数的闭包变量是否发生变更，如果变更了也要进行更新：
    • 对于带有 super () 调用的函数，其闭包内会存储自身的引用，这个不需要更新：
    class c_1():
        def __init__(self):
            super().__init__()
            
    >>> print(c_1.__init__.__code__.co_freevars[0])
    __class__
    >>> print(c_1.__init__.__closure__[0].cell_contents)
    <class '__main__.c_1'>
  • 闭包参数数量不一致的情况下，直接替换，更新意义不大。
def reload_func(old_func:types.FunctionType, new_func:types.FunctionType, depth = 0):
    """
    热更函数(func):
        ①: 更新旧函数里面的属性，详细内容见下方定义
        ②: 部分属性不可写或无法修改的不做处理
        ③: 处理闭包 cellvar
        ④: 新增函数或变量直接添加
        ⑤: 类型变更直接替换
        ⑥: 其他情况也直接替换即可
    函数基本定义:
    class FunctionType:
        __closure__: Optional[Tuple[_Cell, ...]]
        __code__: CodeType
        __defaults__: Optional[Tuple[Any, ...]]
        __dict__: Dict[str, Any]
        __globals__: Dict[str, Any]
        __name__: str
        __qualname__: str
        __annotations__: Dict[str, Any]
        __kwdefaults__: Dict[str, Any]
        def __init__(self, code: CodeType, globals: Dict[str, Any], name: Optional[str] = ..., argdefs: Optional[Tuple[object, ...]] = ..., closure: Optional[Tuple[_Cell, ...]] = ...) -> None: ...
        def __call__(self, *args: Any, **kwargs: Any) -> Any: ...
        def __get__(self, obj: Optional[object], type: Optional[type]) -> MethodType: ...
    闭包问题:
        - Python 函数调用时出现闭包参数不一致,热更后会导致报错
            def out_func():
                arg1 = 1
                def inner_func():
                    print(arg1)
                return inner_func
            
            f = out_func # 热更前代码
            f = out_func # 热更后代码
            f() # 调用
            # out: requires a code object with 1 free vars, not 0
    闭包原理:
        - out_func 在执行过程中，会把自身运行栈中内层函数引用的变量以 ob_ref 的形式绑定到 co_cellvars 的tuple当中(out_func.__closure__)。
        - 在 inner_func 对象内，解开传递进来的 co_cellvars 的tuple并重新绑定到自己的 co_freevars 的tuple中
        - 如果想要热更闭包内容，只需要替换掉 inner_func.__closure__ 里的内容
    部分不可变属性:
        - "__class__"   assignment only supported for heap types or ModuleType subclasses
        - "__closure__" readonly attribute
        - "__globals__" readonly attribute
    """
   
    if depth > 2:
        return False
    
    # 闭包参数不一致，无法更新
    old_cell_var_num = len(old_func.__closure__) if old_func.__closure__ else 0
    new_cell_var_num = len(new_func.__closure__) if new_func.__closure__ else 0
    if old_cell_var_num != new_cell_var_num:
        return False
    # 更新属性
    setattr(old_func, '__code__', new_func.__code__)
    setattr(old_func, '__defaults__', new_func.__defaults__)
    setattr(old_func, '__dict__', new_func.__dict__)
    setattr(old_func, '__name__', new_func.__name__)
    setattr(old_func, '__qualname__', new_func.__qualname__)
    setattr(old_func, '__annotations__', new_func.__annotations__)
    # 类型标注
    setattr(old_func, '__kwdefaults__', new_func.__kwdefaults__)
    # def m(cls,a=1,b=2,*kwarg,g=1,v=2): __kwdefaults__ = {'g': 1, 'v': 2}
    
    # 更新闭包参数
    if old_cell_var_num > 0:
        ignore_idx = []
        for idx, freevar in enumerate(old_func.__code__.co_freevars):
            # super () 操作不改变其指向的父类 __class__
            if freevar == "__class__":
                ignore_idx.append(idx)
        for idx, old_cellvar in enumerate(old_func.__closure__):
            if idx in ignore_idx:
                continue
            new_cellvar = new_func.__closure__[idx]
            # 闭包参数是函数的话递归更新
            if inspect.isfunction(old_cellvar.cell_contents) and inspect.isfunction(new_cellvar.cell_contents):
                if not reload_func(old_cellvar.cell_contents, new_cellvar.cell_contents, depth + 1):
                    old_cellvar.cell_contents = new_cellvar.cell_contents
            # 其他情况都视作替换
            else:
                old_cellvar.cell_contents = new_cellvar.cell_contents
    return True
# 热更 global object：
全局对象的更新就比较的简单，大致可以分为两种：
  • 能够获取到所有需要热更对象的引用，可以直接在对象上进行修改。
  • 不能获取的情况下，直接修改模块内的定义。缺点就是已经创建了的对象，内容还是旧的。
# 测试代码：
  • 修改前的代码：
# -*- coding: utf8 -*-
import ccore
import enum
import functools
import dataclasses
import typing
import collections
# 全局变量测试
a_1 = 1
a_2 = [1,2]
a_3 = (1,2,3)
a_4 = {1:1,2:2}
a_5 = {1,2,3}
a_6 = 1.1
a_7 = object
a_8 = len
a_9 = functools.partial(len, a_2)
a_10 = collections.defaultdict(int)
# 函数测试
def b_1(arg1, arg2):
    b_100 = 1
    b_200 = 1
    return b_100+b_200
def b_2(arg1, arg2):
    return arg1 + arg2
def b_3(arg1, arg2, *args):
    return arg2 + arg1 + sum(args)
def b_4(arg1 = 1, arg2 = 2, *args):
    return arg2 + arg1 + sum(args)
def b_5(arg1 = 1, arg2 = 2, *args, **kwargs):
    return arg2 + arg1 + sum(args) + sum(kwargs.values())
def b_6(arg1 = 1, arg2 = 2, *args, arg3 = 4, arg4 = 3):
    return arg2 + arg1 + sum(args) + arg3 + arg4
def b_7(func, arg1 = 1, arg2 = 2):
    @functools.wraps(func)
    def inner(*args, **kwargs):
        return arg1 + arg2 + func()
    return inner
@b_7
def b_8(arg1 = 1, arg2 = 2, *args, arg3 = 4, arg4 = 3):
    return arg2 + arg1 + sum(args) + arg3 + arg4
def b_9(arg1 = 1, arg2 = 2):
    def wapper(func):
        @functools.wraps(func)
        def inner(*args, **kwargs):
            return arg1 + arg2 + func()
        return inner
    return wapper
@b_9(1,2)
def b_10(arg1 = 1, arg2 = 2, *args, arg3 = 4, arg4 = 3):
    return arg2 + arg1 + sum(args) + arg3 + arg4
# 类测试
class c_1():
    c_100 = 1
    c_200 = 2
    def __init__(self):
        super().__init__()
    def __call__(self):
        return False
    def c_1000(self,arg1 = 1, arg2 = 2, *args, arg3 = 4, arg4 = 3):
        return arg2 + arg1 + sum(args) + arg3 + arg4
    @property
    def C_100(self):
        return self.c_100
    @classmethod
    def c_2000(cls,arg1 = 1, arg2 = 2, *args, arg3 = 4, arg4 = 3):
        return arg2 + arg1 + sum(args) + arg3 + arg4
    @staticmethod
    def c_3000(arg1 = 1, arg2 = 2, *args, arg3 = 4, arg4 = 3):
        return arg2 + arg1 + sum(args) + arg3 + arg4
    def c_4000(self, arg1 = 1, arg2 = 2):
        def wapper(func):
            @functools.wraps(func)
            def inner(*args, **kwargs):
                ...
            return inner
        return wapper
@dataclasses.dataclass
class d_1:
    d_100:int = 1
    d_200:str = "1"
    d_300:typing.List[int] = dataclasses.field(default_factory=list)
    d_400:typing.Dict[int,int] = dataclasses.field(default_factory=dict)
e_1 = collections.namedtuple("e_1", "e_100 e_200 e_300 e_400")
class INFO(enum.IntEnum):
    m_1 = 1
    m_2 = 2
    m_3 = 3
  • 修改后的代码：
# -*- coding: utf8 -*-
import ccore
import enum
import functools
import dataclasses
import typing
import collections
# 全局变量测试
a_1 = 2
a_2 = [2,2]
a_3 = (2,2,3)
a_4 = {2:1,2:2}
a_5 = {2,2,3}
a_6 = 2.1
a_7 = object
a_8 = len
a_9 = functools.partial(len, a_3)
a_10 = collections.defaultdict(str)
# 函数测试
def b_1(arg1):
    b_100 = 3
    b_200 = 4
    return b_100+b_200
def b_2(arg1):
    return arg1
def b_3(arg1, *args):
    return arg1 + sum(args)
def b_4(arg1 = 1, *args):
    return arg1 + sum(args)
def b_5(arg1 = 1, *args, **kwargs):
    return arg1 + sum(args) + sum(kwargs.values())
def b_6(arg1 = 1, *args, arg3 = 4, arg4 = 3):
    return arg1 + sum(args) + arg3 + arg4
def b_7(func, arg1 = 1):
    @functools.wraps(func)
    def inner(*args, **kwargs):
        return arg1 + func()
    return inner
@b_7
def b_8(arg1 = 1, *args, arg3 = 4, arg4 = 3):
    return arg1 + sum(args) + arg3 + arg4
def b_9(arg1 = 1):
    def wapper(func):
        @functools.wraps(func)
        def inner(*args, **kwargs):
            return arg1+ func()
        return inner
    return wapper
@b_9(1)
def b_10(arg1 = 1, *args, arg3 = 4, arg4 = 3):
    return arg1 + sum(args) + arg3 + arg4
# 类测试
class c_1():
    c_100 = 10
    c_200 = 20
    def __init__(self):
        super().__init__()
    def __call__(self):
        return False
    def c_1000(self,arg1 = 1, *args, arg3 = 4, arg4 = 3):
        return arg1 + sum(args) + arg3 + arg4
    @property
    def C_100(self):
        return self.c_100
    @classmethod
    def c_2000(cls,arg1 = 1, *args, arg3 = 4, arg4 = 3):
        return arg1 + sum(args) + arg3 + arg4
    @staticmethod
    def c_3000(arg1 = 1, *args, arg3 = 4, arg4 = 3):
        return arg1 + sum(args) + arg3 + arg4
    def c_4000(self, arg1 = 1):
        def wapper(func):
            @functools.wraps(func)
            def inner(*args, **kwargs):
                ...
            return inner
        return wapper
@dataclasses.dataclass
class d_1:
    d_100:int = 2
    d_200:str = "2"
    d_300:int = 1
    d_400:str = "1"
e_1 = collections.namedtuple("e_1", "e_100 e_200 e_300")
class INFO(enum.IntEnum):
    m_1 = 3
    m_2 = 2
    m_3 = 1
  • 测试数据初始化（1 表示热更前，2 表示热更后）
def init_data(flag=1):
    import collections
    d = collections.OrderedDict()
    d["data"] = {
        "a_1" : fix_module.a_1,
        "a_2" : fix_module.a_2,
        "a_3" : fix_module.a_3,
        "a_4" : fix_module.a_4,
        "a_5" : fix_module.a_5,
        "a_6" : fix_module.a_6,
        "a_7" : fix_module.a_7,
        "a_8" : fix_module.a_8,
        "a_9" : fix_module.a_9,
        "a_10" : fix_module.a_10,
    }
    d["func"] = {
        "b_1": fix_module.b_1,
        "b_2": fix_module.b_2,
        "b_3": fix_module.b_3,
        "b_4": fix_module.b_4,
        "b_5": fix_module.b_5,
        "b_6": fix_module.b_6,
        "b_7": fix_module.b_7,
        "b_8": fix_module.b_8,
        "b_9": fix_module.b_9,
        "b_10": fix_module.b_10,
    }
    if flag == 1:
        d["class"] = {
            "c_1":fix_module.c_1(),
            "d_1":fix_module.d_1(d_100=1,d_200="1",d_300=[1,],d_400={1:1}),
            "e_1":fix_module.e_1(e_100 = 1,e_200 =2,e_300 =3,e_400 = 4)
        }
    else:
        d["class"] = {
            "c_1":fix_module.c_1(),
            "d_1":fix_module.d_1(d_100=1,d_200="1",d_300=2,d_400="2"),
            "e_1":fix_module.e_1(e_100 = 1,e_200 =2,e_300 =3)
        }  
    return d
def Output(d, flag=1):
     
    for k,v in d["data"].items():
        print(f" name {k} val {v}")
    if flag == 1:
        for k,v in d["func"].items():
            print(f" name {k} val {v(1,2)}")
        c_1 = d["class"]["c_1"]
        print(f"c_1.c_100 {c_1.c_100}")
        print(f"c_1.c_200 {c_1.c_200}")
        print(f"c_1.c_1000(1,2) {c_1.c_1000(1,2)}")
        print(f"c_1.C_100 {c_1.C_100}")
        print(f"c_1.c_2000(1,2) {c_1.c_2000(1,2)}")
        print(f"c_1.c_3000(1,2) {c_1.c_3000(1,2)}")
        print(f"c_1.c_4000(1,2) {c_1.c_4000(1,2)}")
    else:
        for k,v in d["func"].items():
            print(f" name {k} val {v(1)}")
        c_1 = d["class"]["c_1"]
        print(f"c_1.c_100 {c_1.c_100}")
        print(f"c_1.c_200 {c_1.c_200}")
        print(f"c_1.c_1000(1) {c_1.c_1000(1)}")
        print(f"c_1.C_100 {c_1.C_100}")
        print(f"c_1.c_2000(1) {c_1.c_2000(1)}")
        print(f"c_1.c_3000(1) {c_1.c_3000(1)}")
        print(f"c_1.c_4000(1) {c_1.c_4000(1)}")
    
    d_1 = d["class"]["d_1"]
    print(f"d_1.d_100 {d_1.d_100}")
    print(f"d_1.d_200 {d_1.d_200}")
    print(f"d_1.d_300 {d_1.d_300}")
    print(f"d_1.d_400 {d_1.d_400}")
    e_1 = d["class"]["e_1"]
    print(f"e_1.e_100 {e_1.e_100}")
    print(f"e_1.e_200 {e_1.e_200}")
    print(f"e_1.e_300 {e_1.e_300}")
    if flag == 1:
        print(f"e_1.e_400 {e_1.e_400}")
    print(fix_module.INFO.m_1.value)
    print(fix_module.INFO.m_2.value)
    print(fix_module.INFO.m_3.value)
  • 最终的输出结果
#------------------------------------------------------- 旧模块热更前的输出：-------------------------------------------------------
name a_1 val 1
name a_2 val [1, 2]
name a_3 val (1, 2, 3)
name a_4 val {1: 1, 2: 2}
name a_5 val {1, 2, 3}
name a_6 val 1.1
name a_7 val <class 'object'>
name a_8 val <built-in function len>
name a_9 val functools.partial(<built-in function len>, [1, 2])
name a_10 val defaultdict(<class 'int'>, {})
name b_1 val 2
name b_2 val 3
name b_3 val 3
name b_4 val 3
name b_5 val 3
name b_6 val 10
name b_7 val <function b_7.<locals>.inner at 0x000002E77A993700>
name b_8 val 13
name b_9 val <function b_9.<locals>.wapper at 0x000002E77A993700>
name b_10 val 13
c_1.c_100 1
c_1.c_200 2
c_1.c_1000(1,2) 10
c_1.C_100 1
c_1.c_2000(1,2) 10
c_1.c_3000(1,2) 10
c_1.c_4000(1,2) <function c_1.c_4000.<locals>.wapper at 0x000002E77A993700>
d_1.d_100 1
d_1.d_200 1
d_1.d_300 [1]
d_1.d_400 {1: 1}
e_1.e_100 1
e_1.e_200 2
e_1.e_300 3
e_1.e_400 4
1
2
3
#------------------------------------------------------- 旧模块热更后的输出：-------------------------------------------------------
name a_1 val 2
name a_2 val [2, 2]
name a_3 val (2, 2, 3)
name a_4 val {2: 2}
name a_5 val {2, 3}
name a_6 val 2.1
name a_7 val <class 'object'>
name a_8 val <built-in function len>
name a_9 val functools.partial(<built-in function len>, (2, 2, 3))
name a_10 val defaultdict(<class 'str'>, {})
name b_1 val 7
name b_2 val 1
name b_3 val 1
name b_4 val 1
name b_5 val 1
name b_6 val 8
name b_7 val <function b_7.<locals>.inner at 0x000002E77A9905E0>
name b_8 val 9
name b_9 val <function b_9.<locals>.wapper at 0x000002E77A9905E0>
name b_10 val 9
c_1.c_100 10
c_1.c_200 20
c_1.c_1000(1) 8
c_1.C_100 10
c_1.c_2000(1) 8
c_1.c_3000(1) 8
c_1.c_4000(1) <function c_1.c_4000.<locals>.wapper at 0x000002E77A9905E0>
d_1.d_100 1
d_1.d_200 1
d_1.d_300 2
d_1.d_400 2
e_1.e_100 1
e_1.e_200 2
e_1.e_300 3
3
2
1
 
 更新于   阅读次数  次
请我[恰饭]~(￣▽￣)~*
鑫酱(●'◡'●) 微信支付
微信支付
鑫酱(●'◡'●) 支付宝
支付宝