python - 作为库开发人员，如何防止用户跨多个线程/进程使用对象？

Question

我正在为 Python 构建一个库，并注意到我的一个用户通过在 2 个进程之间共享对象而错误地使用了这些对象。

所以，我想以一种防止用户这样做的方式进行烘焙。

一种方法是使用装饰器在每个方法调用之前添加“检查”，但这会在每个方法调用上增加一些开销。

import multiprocessing
import threading


def is_parent():
    return (
        multiprocessing.current_process().name == "MainProcess"
        and threading.current_thread() == threading.main_thread()
    )


def process_unsafe(fn):
    def wrapper(*args, **kwargs):
        if not is_parent():
            raise RuntimeError("Not allowed!")
        fn(*args, **kwargs)

    return wrapper


class NonProcessSafe:
    @process_unsafe
    def foo(self):
        pass

    @process_unsafe
    def bar(self):
        pass


nps = NonProcessSafe()

nps.foo()
nps.bar()


def child():
    # complains
    nps.foo()


p = multiprocessing.Process(target=child)
p.start()
p.join()

有没有更有效的方法来实现这一目标？

Answer 1

正如 @timgeb 在评论中提到的，可能没有办法堵住所有漏洞。此外，我不确定我是否完全理解您的问题和进程/线程/状态共享模型......

无论如何，这里的 Ansatz 可能会让你有所帮助，至少在进程方面(在 Unix/Linux 意义上):

import os

class State:

    def __init__(self):
        self._pid = os.getpid()
        self._state = {}

    def __setitem__(self, key, value):
        if os.getpid() != self._pid:
            raise RuntimeError("this process must not change state")
        self._state[key] = value

State 实例会记住创建它的进程的 ID，然后可以在所有(相关)方法中检查该 ID。这当然要求状态由其“所有者”进程创建，在这种情况下，它对任何父进程或兄弟进程都是不可见的，并且仅对其自身及其子进程可见。

对于线程，使用threading.get_ident()而不是os.getpid()。

如果始终是“主”进程创建状态，然后 fork 工作子进程，那么您可以(在主进程中)暂时将状态标记为只读， fork 工作子进程，以便它获得只读状态只复制，然后在主进程中切换状态回读写:

class State:
    def __init__(self):
        self._read_only = False
        self._state = {}
    def __setattr__(self, key, value):
        if self._read_only:
            raise RuntimeError("read only!")
        self._state[key] = value

在主进程中:

def new_worker(fct):
    state = State()
    state._read_only = True
    pid = os.fork()  # or whatever you use to create a new process
    if pid:
        fct()  # state is _read-only=True here in child process
    else:
        state._read_only = False
        state_list.append((pid, state))