Как сделать встроенные контейнеры (наборы, дикты, списки) безопасными потоками?

Ответ 1

Вы можете использовать средства метапрограммирования Python для этого. (Примечание: написано быстро и не проверено полностью.) Я предпочитаю использовать декоратор класса.

Я также думаю, что вам может потребоваться заблокировать более add и remove, чтобы сделать набор потокобезопасным, но я не уверен. Я проигнорирую эту проблему и сосредоточусь только на вашем вопросе.

Также рассмотрите, является ли делегирование (проксирование) более подходящим, чем подкласс. Обтекание объектов - это обычный подход в Python.

Наконец, нет "волшебной палочки" метапрограммирования, которая волшебным образом добавит мелкозернистую блокировку к любой изменяемой коллекции Python. Самое безопасное, что нужно сделать, это заблокировать доступ к любому методу или атрибуту с помощью RLock, но это очень грубое и медленное и, вероятно, все еще не гарантирует, что ваш объект будет потокобезопасным во всех случаях. (Например, у вас может быть коллекция, которая управляет другим объектом, не связанным с потоками, доступным для других потоков.) Вам действительно нужно изучить каждую структуру данных и подумать о том, какие операции являются атомарными или требуют блокировок и какие методы могут вызывать другие методы используя ту же самую блокировку (то есть, тупик).

Тем не менее, вот некоторые методы в вашем распоряжении в порядке увеличения абстракции:

Делегирование

class LockProxy(object):
    def __init__(self, obj):
        self.__obj = obj
        self.__lock = RLock()
        # RLock because object methods may call own methods
    def __getattr__(self, name):
        def wrapped(*a, **k):
            with self.__lock:
                getattr(self.__obj, name)(*a, **k)
        return wrapped

lockedset = LockProxy(set([1,2,3]))

Контекстный менеджер

class LockedSet(set):
    """A set where add(), remove(), and 'in' operator are thread-safe"""

    def __init__(self, *args, **kwargs):
        self._lock = Lock()
        super(LockedSet, self).__init__(*args, **kwargs)

    def add(self, elem):
        with self._lock:
            super(LockedSet, self).add(elem)

    def remove(self, elem):
        with self._lock:
            super(LockedSet, self).remove(elem)

    def __contains__(self, elem):
        with self._lock:
            super(LockedSet, self).__contains__(elem)

декоратор

def locked_method(method):
    """Method decorator. Requires a lock object at self._lock"""
    def newmethod(self, *args, **kwargs):
        with self._lock:
            return method(self, *args, **kwargs)
    return newmethod

class DecoratorLockedSet(set):
    def __init__(self, *args, **kwargs):
        self._lock = Lock()
        super(DecoratorLockedSet, self).__init__(*args, **kwargs)

    @locked_method
    def add(self, *args, **kwargs):
        return super(DecoratorLockedSet, self).add(elem)

    @locked_method
    def remove(self, *args, **kwargs):
        return super(DecoratorLockedSet, self).remove(elem)

Декодер класса

Я думаю, что это самый чистый и простой в понимании абстрактных методов, поэтому я расширил его, чтобы можно было указать методы блокировки и объект блокировки factory.

def lock_class(methodnames, lockfactory):
    return lambda cls: make_threadsafe(cls, methodnames, lockfactory)

def lock_method(method):
    if getattr(method, '__is_locked', False):
        raise TypeError("Method %r is already locked!" % method)
    def locked_method(self, *arg, **kwarg):
        with self._lock:
            return method(self, *arg, **kwarg)
    locked_method.__name__ = '%s(%s)' % ('lock_method', method.__name__)
    locked_method.__is_locked = True
    return locked_method


def make_threadsafe(cls, methodnames, lockfactory):
    init = cls.__init__
    def newinit(self, *arg, **kwarg):
        init(self, *arg, **kwarg)
        self._lock = lockfactory()
    cls.__init__ = newinit

    for methodname in methodnames:
        oldmethod = getattr(cls, methodname)
        newmethod = lock_method(oldmethod)
        setattr(cls, methodname, newmethod)

    return cls


@lock_class(['add','remove'], Lock)
class ClassDecoratorLockedSet(set):
    @lock_method # if you double-lock a method, a TypeError is raised
    def frobnify(self):
        pass

Отменить доступ к атрибуту с помощью __getattribute__

class AttrLockedSet(set):
    def __init__(self, *args, **kwargs):
        self._lock = Lock()
        super(AttrLockedSet, self).__init__(*args, **kwargs)

    def __getattribute__(self, name):
        if name in ['add','remove']:
            # note: makes a new callable object "lockedmethod" on every call
            # best to add a layer of memoization
            lock = self._lock
            def lockedmethod(*args, **kwargs):
                with lock:
                    return super(AttrLockedSet, self).__getattribute__(name)(*args, **kwargs)
            return lockedmethod
        else:
            return super(AttrLockedSet, self).__getattribute__(name)

Динамически добавленные методы-обертки с __new__

class NewLockedSet(set):
    def __new__(cls, *args, **kwargs):
        # modify the class by adding new unbound methods
        # you could also attach a single __getattribute__ like above
        for membername in ['add', 'remove']:
            def scoper(membername=membername):
                # You can also return the function or use a class
                def lockedmethod(self, *args, **kwargs):
                    with self._lock:
                        m = getattr(super(NewLockedSet, self), membername)
                        return m(*args, **kwargs)
                lockedmethod.__name__ = membername
                setattr(cls, membername, lockedmethod)
        self = super(NewLockedSet, cls).__new__(cls, *args, **kwargs)
        self._lock = Lock()
        return self

Динамически добавленные методы-обертки с __metaclass__

def _lockname(classname):
    return '_%s__%s' % (classname, 'lock')

class LockedClass(type):
    def __new__(mcls, name, bases, dict_):
        # we'll bind these after we add the methods
        cls = None
        def lockmethodfactory(methodname, lockattr):
            def lockedmethod(self, *args, **kwargs):
                with getattr(self, lockattr):
                    m = getattr(super(cls, self), methodname)
                    return m(*args,**kwargs)
            lockedmethod.__name__ = methodname
            return lockedmethod
        lockattr = _lockname(name)
        for methodname in ['add','remove']:
            dict_[methodname] = lockmethodfactory(methodname, lockattr)
        cls = type.__new__(mcls, name, bases, dict_)
        return cls

    def __call__(self, *args, **kwargs):
        #self is a class--i.e. an "instance" of the LockedClass type
        instance = super(LockedClass, self).__call__(*args, **kwargs)
        setattr(instance, _lockname(self.__name__), Lock())
        return instance



class MetaLockedSet(set):
    __metaclass__ = LockedClass

Динамически созданные метаклассы

def LockedClassMetaFactory(wrapmethods):
    class LockedClass(type):
        def __new__(mcls, name, bases, dict_):
            # we'll bind these after we add the methods
            cls = None
            def lockmethodfactory(methodname, lockattr):
                def lockedmethod(self, *args, **kwargs):
                    with getattr(self, lockattr):
                        m = getattr(super(cls, self), methodname)
                        return m(*args,**kwargs)
                lockedmethod.__name__ = methodname
                return lockedmethod
            lockattr = _lockname(name)
            for methodname in wrapmethods:
                dict_[methodname] = lockmethodfactory(methodname, lockattr)
            cls = type.__new__(mcls, name, bases, dict_)
            return cls

        def __call__(self, *args, **kwargs):
            #self is a class--i.e. an "instance" of the LockedClass type
            instance = super(LockedClass, self).__call__(*args, **kwargs)
            setattr(instance, _lockname(self.__name__), Lock())
            return instance
    return LockedClass

class MetaFactoryLockedSet(set):
    __metaclass__ = LockedClassMetaFactory(['add','remove'])

Я ставлю, используя простой, явный try...finally теперь не выглядит так плохо, не так ли?

Упражнение для читателя: пусть вызывающий передаёт свой собственный объект Lock() (инъекция зависимостей) с помощью любого из этих методов.

Ответ 2

[Действительно, см. комментарии, это не так)

Если вы используете CPython, вы можете увидеть из исходного кода, что он не выпускает GIL (http://hg.python.org/cpython/file/db20367b20de/Objects/setobject.c), поэтому все его операции должны быть атомарными.

Если это все, что вам нужно, и вы обязательно запустите свой код на CPython, вы можете просто использовать его напрямую.

Ответ 3

Это моя первая попытка играть с декораторами (хотя мой код фактически не использует синтаксис @decorate), и у меня нет большого опыта многопоточности/многопроцессорности. Однако с этой оговоркой я попытался сделать следующее:

from multiprocessing import Lock

def decorate_all(obj):
    lock = Lock()
    #you'll want to make this more robust:
    fnc_names = [fnctn for fnctn in dir(obj) if '__' not in fnctn]
    for name in fnc_names:
        print 'decorating ' + name
        fnc = getattr(obj, name)
        setattr(obj, name, decorate(fnc, lock))
    return obj

def decorate(fnctn, lock):
    def decorated(*args):
        print 'acquiring lock'
        lock.acquire()
        try:
            print 'calling decorated function'
            return fnctn(*args)
        finally:
            print 'releasing lock'
            lock.release()
    return decorated


def thread_safe(superclass):
    lock = Lock()
    class Thread_Safe(superclass):
        def __init__(self, *args, **kwargs):
            super(Thread_Safe, self).__init__(*args, **kwargs)
    return decorate_all(Thread_Safe)


>>> thread_safe_set = thread_safe(set)
decorating add
decorating clear
decorating copy
decorating difference
decorating difference_update
decorating discard
decorating intersection
decorating intersection_update
decorating isdisjoint
decorating issubset
decorating issuperset
decorating pop
decorating remove
decorating symmetric_difference
decorating symmetric_difference_update
decorating union
decorating update
>>> s = thread_safe_set()
>>> s.add(1)
acquiring lock
calling decorated function
releasing lock
>>> s.add(4)
acquiring lock
calling decorated function
releasing lock
>>> s.pop()
acquiring lock
calling decorated function
releasing lock
1
>>> s.pop()
acquiring lock
calling decorated function
releasing lock
4
>>>

Ответ 4

Вы можете реализовать свой собственный менеджер контекстов:

class LockableSet:
    def __enter__(self):
        self.lock()
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        #Do what you want with the error
        self.unlock()

with LockableSet() as s:
    s.whatever()
    raise Exception()

Независимо от того, в конце будет вызываться метод __exit__. Более подробная информация доступна здесь (официальные документы python).

Другим использованием для этого может быть декоратор lock для методов, например:

def lock(func):
    def safe_func(self, *args, **kwargs):
        with self:
            func(self, *args, **kwargs)
    return safe_func