Блог

Мы пытаемся использовать метаклассы для пользовательского внутреннего выбора ( multiprocessing.Process или threading.Thread ). Обоснование этой реализации заключается в расширении функциональных возможностей процесса/потока для нашего пользовательского использования. В то время как следующий код работает для fork (по умолчанию в unix). Но для spawn (по умолчанию в Windows) я получаю ошибку.

 pickle.PicklingError: Can't pickle <class '__main__.DProcess'>: it's not the same object as __main__.DProcess
 

Ошибка возникает из модуля рассола, так как объекты не совпадают.

 obj: <class '__main__.DProcess'>, 
obj.__dict__: {'__module__': '__main__', 'run': <function DProcess.run at 0x7fa76ccd97a0>, '__doc__': None, '__slotnames__': []}
hash(obj): 5875061359185

obj2: <class '__main__.DProcess'>, 
obj2.__dict__: {'__module__': '__main__', 'run': <function DProcess.run at 0x7fa76ccd97a0>, '__dict__': <attribute '__dict__' of 'DProcess' objects>, '__weakref__': <attribute '__weakref__' of 'DProcess' objects>, '__doc__': None}, 
hash(obj2): 5875061305336
 

Я не совсем понимаю, что здесь происходит.

1. Почему эти 2 объекта разные? Выполнение save_global из модуля pickle для объекта класса не завершается ошибкой. Это из-за __call__ реализации? Как мне это исправить?
2. Почему эта проверка не выполняется для вилки?

Вот код:

 class Backend(type):
    _cache = {}

    def __new__(cls, name, bases, dct):
        _cls = super().__new__(cls, name, bases, dct)
        # store the subclass dict to be used during __call__
        Backend._cache.update(
            {name: {'cls': cls, 'name': name, 'bases': bases, 'dct': dct}}
        )
        return _cls

    def __call__(cls, *args, **kwargs) -> 'Backend':
        try:
            # check arg amp; select the base class
            if args[0] == 'process':
                import multiprocessing
                _cls = multiprocessing.Process
            elif args[0] == 'thread':
                import threading
                _cls = threading.Thread
        except KeyError:
            print('Please pass process or thread as the 1st arg')

        for c in cls.mro()[-2::-1]:
            # pick args from __new__ and call type()
            arg_cls = Backend._cache[c.__name__]['cls']
            arg_name = Backend._cache[c.__name__]['name']
            arg_dct = Backend._cache[c.__name__]['dct']
            _cls = super().__new__(arg_cls, arg_name, (_cls,), arg_dct)

        return type.__call__(_cls, *args[1:], **kwargs)


class DProcess(metaclass=Backend):
    def run(self):
        print('we are in dprocess')
        super().run()


if __name__ == '__main__':
    from multiprocessing import set_start_method as _set_start_method
    _set_start_method('spawn')
    DProcess('process').start()
 

Если вам не нужен метакласс, вы не должны его использовать — есть лучшие шаблоны для того, что вы хотите. Прежде всего: вам действительно нужно наследовать от потока или процесса?? Возможно, лучший выбор — просто иметь их в качестве связанного атрибута в вашем классе DProcess, и тогда он может работать как обычный атрибут класса.

Поскольку важным интерфейсом для обоих в основном является установка целевого вызываемого объекта, start и join вы можете либо создать для них прокси-методы, либо просто вызвать метод непосредственно в атрибуте класса.

Т. е. ваш дизайн, скорее всего, может работать именно так

 

class DProcess():
    def __init__(self, backend):
        if backend == "process":
            self.backend_cls = multiprocessing.Process
        elif backend == "thread":
            self.backend_cls = threading.Thread
            
        self.worker = self.backend_cls(target=self.run)
        
    def start(self):
        
        self.worker.start()
        # or just call "instance.worker.start()" from outside
        
    def join(self):
        return self.worker.join()
    
    def run(self):
        print('we are in dprocess')
        super().run()
        
 

Теперь причины, по которым ваш исходный код терпит неудачу, заключаются в том, что он неправильный: вы на самом деле создаете новый класс-брат для обработки в каждом экземпляре
DПроцесс, динамически, вызывающий super().__new__ метакласс __call__ .

Итак, class DProcess объявленный в вашем ядре один класс. Но каждый раз, когда вы пытаетесь создать его, создается новый класс objct, и он создается — вот на что жалуется пикл. (и пока мы на этом: мультипроцессорная обработка с помощью fork всего имеют точно такие же объекты на новый процесс, в то время как для Windows так нужно запустить новый процесс с нуля, и сериализовать объекты, поэтому они отправляются в новый процесс — это не «проверка»- это то, что «призрак брата» из DProcess не может быть десериализован на огурец, так как он не существует на другой процесс.

Теперь, если вы действительно хотите, чтобы ваши классы наследовались от потока или процесса, вы можете просто создать два класса и использовать функцию фабрики, чтобы выбрать, какой из них вы хотите. В то время как было бы тривиально иметь функцию для создания обоих похожих классов, а затем помещать ее в список или глобальный словарь, Pickle это не очень нравится: ему нужно, чтобы экземпляры или классы, которые нужно замариновать, были объявлены на верхнем уровне модуля (чтобы полное имя класса могло вернуть вас к конструктору класса). И даже там нет необходимости повторять код — вы можете просто использовать класс mixin с любым вашим общим кодом, и с помощью двух строк вы создадите свой ProcessDworker и ThreadDWorker (которые затем могут быть выбраны заводской функцией).:

 
    
    
class Stub:
    """just needed in case some linter or static checker complain about
    these methods not being present in the mixin
    
    But you could also declare these as @abstractmethod 
    to ensure just a proper class incorporating Thread or Process can
    be instantiated
    """
    
    def run(self): pass
    def start(self): pass
    def join(self): pass

class DProcessMixin(Stub):
    def __init__(self, *args, **kw):
        # whatever code you need to setup yoru worker - like creating queues, and such
        ...
        super().__init__(self, ...)
    
    ...
    
class ThreadDprocess(DProcessMixin, threading.Thread):
    queue_class = threading.Queue
    pass

class ProcessDProcess(DProcessMixin, threadng.
    queue_class = multiprocessing.Queue
    pass


def DProcess(*args, backend, **kwargs):
    if backend == "process":
        cls = ProcessDProcess
    elif backend == "thread":
        cls = ThreadDprocess
    return cls(*args, **kwargs)
 

И, наконец, если вы действительно хотите запустить метакласс, просто поймите, что __call__ метод в метаклассе находится в том же положении, что и эта Dprocess заводская функция в последнем примере. Если вы предварительно создадите оба класса и фактически кэшируете их, а также зададите им обоим реальное имя в модуле globals это сработало бы. Но если вы вернетесь к своему «кэшу», вы увидите, что он поддельный: он даже не может фактически «кэшировать» информацию для более чем одного класса в одном и том же метаклассе: в вашем кэше должно быть имя класса в качестве ключа, а в качестве значения у вас может быть дополнительный словарь, содержащий значение «имя, базы, пространство имен» для каждого класса. Кстати, вы также путаете cls arg, переданный в метакласс __new__ с самим классом — это тоже неправильно. Короче говоря: я не думаю, что вы достаточно разбираетесь в работе механизма классов, чтобы строить код вокруг этого, и поскольку ваша проблема, по-видимому, тривиально решается только с помощью композиции, это должно быть бесполезно.