如何制作2.7 python上下文管理器线程安全

我有一个运行在Django服务上的大型python应用程序。我需要关闭许可测试对于某些操作，所以我创造了这个情况管理器：使用上下文管理如何制作2.7 python上下文管理器线程安全

class OverrideTests(object): 

    def __init__(self): 
     self.override = 0 

    def __enter__(self): 
     self.override += 1 

    # noinspection PyUnusedLocal 
    def __exit__(self, exc_type, exc_val, exc_tb): 
     self.override -= 1 
     assert not self.override < 0 

    @property 
    def overriding(self): 
     return self.override > 0 

override_tests = OverrideTests()

应用程序的各个部分，然后可以在此改变测试：

with override_tests: 
    do stuff 
    ...

内做的东西，上述上下文管理器可以在不同的功能中多次使用。计数器的使用保持这种控制，并且似乎工作正常......直到线程卷入。

一旦涉及到线程，就会重新使用全局上下文管理器，因此测试可能会错误地被覆盖。

下面是一个简单的测试用例 - 这工作得很好，如果thread.start_new_thread(do_id,())线被替换为简单do_it但引人注目的失败，如下所示：

def stat(k, expected): 
    x = '.' if override_tests.overriding == expected else '*' 
    sys.stdout.write('{0}{1}'.format(k, x)) 


def do_it_inner(): 
    with override_tests: 
     stat(2, True) 
    stat(3, True) # outer with context makes this true 


def do_it(): 
    with override_tests: 
     stat(1, True) 
     do_it_inner() 
    stat(4, False) 


def do_it_lots(ntimes=10): 
    for i in range(ntimes): 
     thread.start_new_thread(do_it,())

我怎样才能让这种情况下管理器线程安全的，这样在每个Python线程，即使它可以重入也一直使用它？

来源

2015-10-29 Paul Whipp

您是否试过不创建单个上下文管理器，而是使用'with OverrideTests（）'，为每个用法创建一个单独的实例？ – BrenBarn

测试代码访问上下文管理器单例中的'全局值'以知道是否应用测试。这不是在与声明的直接背景下。在测试方法中有'if override_tests.overriding：return True'，以便它们被覆盖。如果有多个上下文管理器实例，我不能这么做。 –

任何依赖变异全局状态的东西都会遇到线程问题。听起来你可能需要重新思考这里的整个结构。具有通过检查这种全局标志来改变它们的行为的功能并不是非常健壮的。例如，您可以将所有测试函数放入一个类中，以便每个实例存储自己的覆盖状态，然后为每个线程创建该类的新实例。 – BrenBarn

以下是似乎的工作方式：使您的OverrideTests类为threading.local的子类。为了安全，你应该再调用你__init__超__init__（尽管它似乎即使你不工作）：

class OverrideTests(threading.local): 

    def __init__(self): 
     super(OverrideTests, self).__init__() 
     self.override = 0 

    # rest of class same as before 

override_tests = OverrideTests()

然后：

>>> do_it_lots() 
1.1.1.2.2.1.1.1.1.1.1.3.3.2.2.2.2.2.2.4.4.3.1.3.3.3.3.4.3.2.4.4.2.4.3.4.4.4.3.4.

不过，我不会把在这种情况下，这些钱不会失败，特别是如果你的真实应用比你在这里展示的例子更复杂。最终，你真的应该重新考虑你的设计。在你的问题中，你正在关注如何“使上下文管理器线程安全”。但真正的问题不仅在于你的上下文管理器，而且在于你的函数（在你的例子中为stat）。 stat依赖于全局状态（全球override_tests），这在线程环境中本质上是脆弱的。

来源

2015-11-02 05:03:19 BrenBarn

确定 - 看起来您的解决方案目前工作正常。我已将更改提交给我们的QA服务器，并且如果在接下来的24小时内没有看到重新发布，我们将批准此答案。 –

总会有时候需要依赖某个全局状态，并且完全通过功能参数化来传递这种状态是不切实际的。我的真实世界的例子实际上保持与这个例子一样简单，它似乎运作良好。 –

这已经运行了一段时间了，它似乎已经完全解决了这个问题。 –

-1

threading.RLock是一个可以通过同一个线程多次获取的重入锁。它还支持上下文管理协议，因此可以与with语句一起使用。

它有一个所有者字段，表示当前持有锁的线程。私有方法_is_owned告诉调用线程是否拥有该锁。所有者值可用于确定锁是否由当前线程持有，这使得实现简单。

不需要计数器和线程本地存储。如果当前线程不是所有者，则意味着当前线程不锁定锁定，因此它不会覆盖。

import sys 
from threading import RLock 
try: 
    import _thread as thread 
except ImportError: 
    import thread 

from time import sleep 

class OverrideTests(type(RLock())): 

    @property 
    def overriding(self): 
     return self._is_owned() 

override_tests = OverrideTests() 

def stat(k, expected): 
    x = '.' if override_tests.overriding == expected else '*' 
    sys.stdout.write('{0}{1}'.format(k, x)) 
    sys.stdout.flush() 

def do_it_inner(): 
    with override_tests: 
     stat(2, True) 
    stat(3, True) # outer with context makes this true                           


def do_it(): 
    with override_tests: 
     stat(1, True) 
     do_it_inner() 
    stat(4, False) 


def do_it_lots(ntimes=10): 
    for _ in range(ntimes): 
     thread.start_new_thread(do_it,()) 
     random_sleep() 

if __name__ == '__main__': 
    do_it_lots() 
    sleep(2)

来源

2015-11-06 16:51:09

如何制作2.7 python上下文管理器线程安全

回答

相关问题