用于多处理锁的装饰器在运行时崩溃

Question

iam尝试多处理，并尝试使用锁和装饰器轻松使用锁，但在运行时崩溃

def lock_dec(func):
    def wrapper(*args , **kwargs):
        Lock().acquire()
        func(args)
        Lock().release()
    return wrapper

是装饰

@lock_dec
def add_no_lock(total):
    for i in range(100):
        time.sleep(0.01)
        total.value += 5

这是功能

   add_process = Process(target = add_no_lock , args = (total , ))
   add_process.start()

我遇到此错误，但我无法调试代码

Can't pickle local object 'lock_dec.<locals>.wrapper

Answer 1

尝试24小时后编辑和调试IVE，通过使用decorators with arguments

找到解决方案

def loc_dec_parent(*args , **kwargs):
    def lock_dec(func):
        @wraps(func)    
        def wrapper(*arg , **kwarg):
            kwargs['lock'].acquire()
            try:
                func(*arg)
            finally:
                kwargs['lock'].release()
        return wrapper
    return lock_dec 

和功能是

@loc_dec_parent(lock = Lock())
def add_no_lock(total):
    for i in range(100):
        time.sleep(0.01)
        total.value += 5

这对我有用

Answer 2

您最近的一篇文章引起了我的注意。您的解决方案不是理想的选择，因为它不允许任意参数传递给包装函数(现在它不支持关键字参数(。您的装饰函数只需要一个参数，即要使用的锁，您不应该关心它是否作为关键字参数传递。您还可以通过使用上下文管理器来简化代码：

from functools import wraps
from multiprocessing import Lock
def loc_dec_parent(lock=Lock()):
    def lock_dec(func):
        @wraps(func)
        def wrapper(*args , **kwargs):
            with lock:
                func(*args, **kwargs)
        return wrapper
    return lock_dec
the_lock = Lock()
@loc_dec_parent(the_lock)
def foo(*args, **kwargs):
    print('args:')
    for arg in args:
        print('t', arg)
    print('kwargs:')
    for k, v in kwargs.items():
        print('t', k, '->', v)
foo(1, 2, x=3, lock=4)

打印：

args:
         1
         2
kwargs:
         x -> 3
         lock -> 4

但是，当在Windows或任何使用spawn创建新进程的平台下实际使用时，概念上的装饰器仍然存在问题。

from functools import wraps
from multiprocessing import Lock, Process
import time
def loc_dec_parent(lock=Lock()):
    def lock_dec(func):
        @wraps(func)
        def wrapper(*args , **kwargs):
            with lock:
                func(*args, **kwargs)
        return wrapper
    return lock_dec

lock = Lock()
@loc_dec_parent(lock=lock)
def foo():
    for i in range(3):
        time.sleep(1)
        print(i, flush=True)
@loc_dec_parent(lock=lock)
def bar():
    for i in range(3):
        time.sleep(1)
        print(i, flush=True)
if __name__ == '__main__':
    p1 = Process(target=foo)
    p2 = Process(target=bar)
    p1.start()
    p2.start()
    p1.join()
    p2.join()

打印：

0
0
1
1
2
2

锁定不起作用！我们应该看到以下内容正在工作：

0
1
2
0
1
2

这是因为要实现每个新子过程的创建，在新过程的地址空间中启动了一个新的Python解释器，并且在将控件传递给 target 之前，将源从顶部重新执行。Process实例的。这意味着在每个新过程的地址空间中，都将创建一个新的不同的Lock实例，并重新执行装饰器。

主过程应创建一个Lock实例，然后将其作为参数传递到每个过程。这样，您可以确定每个过程都在处理相同的Lock实例。

简而言之，如果您想支持所有平台，则multiprocessor.Lock是这种装饰人员的不良候选者。

update

要模拟Java的同步方法，您应该确保拥有一个由所有装饰功能和方法使用的单个Lock实例。为此，您想将实现的装饰师用作课程。另外，不要忘记包装器函数应返回包装函数/方法返回的任何可能的返回值。

这必须使用fork在平台上运行以创建新过程：

from functools import wraps
from multiprocessing import Lock, Process
import time
class Synchronized():
    the_lock = Lock() # class instance
    def __call__(self, func):
        @wraps(func)
        def decorated(*args, **kwargs):
            with self.the_lock:
                return func(*args, **kwargs)
        return decorated
@Synchronized()
def foo():
    for i in range(3):
        time.sleep(1)
        print(i, flush=True)
class MyClass:
    @Synchronized()
    def bar(self):
        for i in range(3):
            time.sleep(1)
            print(i, flush=True)
if __name__ == '__main__':
    p1 = Process(target=foo)
    p2 = Process(target=MyClass().bar)
    p1.start()
    p2.start()
    p1.join()
    p2.join()