Найти - Пользователи
Полная версия: не маринуется обьект в multiprocessing.Queue
Начало » Python для новичков » не маринуется обьект в multiprocessing.Queue
1
@cckyi_boxxx
Здравствуйте, имеется слудующая архитектура: главный процесс со своей очередью multiprocessing.Queue и дочерний процесс multiprocessing.Process так-же со своей очередью.

Алгоритм такой: дочерний процесс кладет задание для главного в свою очередь и ждет главный не прочитает это задание и выполнив его не положит ответ в свою очередь. Этакая аналогия multiprocessing.Pipe() только на очередях.

Для обмена данными были написаны два нижеприведенных класса, не хотел использовать словари ибо вычисляемые свойства (property) позволяют контролировать правильность входных данных и как следствие меньше дебага. При этом инстанс класса Drinkins успешно передается как из главного потока в дочерний, так из дочернего в главный. А вот инстанс класса Comandor по каким-то причинам из дочернего потока прилетает в главный с дефолтными значениями, причем при помощи print(comandor_instance.command) в дочернем процессе показывает что все данные введены, пихаю его в очередь команд в дочернем, но когда забираю этот инстанс из очереди в главном потоке и так-же вызываю print(comandor_instance.command) то печатаются дефолтные значения. Однако если дочерний процесс пихает в очередь не инстанс класса Commandor, а словарь при помощи метода Comandor.as_dict(), то все данные прилетают правильно, этим конечно можно воспользоваться, но мне все-же хочется разобраться почему имея два настолько похожих класса один отправляется правильно а второй нет?

Вот код обоих классов:
 class Comandor:
    _currcom = {
        'command':0, # main commands
        'method':0,
        'url':'',
        'js_codes':[],
        'com_data':dict()
        }
    __allowed_commands = {
        'nop':0, # no operation
        'read_page':1, # open new page and return this
        'every_page_jsinj':2, # js injects to every page before give his html
        'set_cur_page_loaded_elem':3, 
        'algo_process_info':4 
        }
    __allowed_methods = {
        'read_page_get':0,
        'read_page_post':1,
        'read_page_scroll':2,
        'read_page_js_inj':3
        }
    
    def __init__(self, **kwargs):
        self.configure(**kwargs)
    
    def as_dict(self):
        ''' written for debug '''
        return self._currcom.copy()
    
    def configure(self, **kwargs):
        '''
        for multiple attribute setting
        '''
        for key, val in kwargs.items():
            if key == 'command': self._s_command(val)
            elif key == 'method': self._s_method(val)
            elif key == 'url': self._s_url(val)
            elif key == 'js_codes': self._s_js_codes(val)
            elif key == 'com_data': self._s_com_data(val)
        return self
    
    ##### command property
    def _g_command(self):
        return self._currcom['command']
    def _s_command(self, data):
        if data in self.__allowed_commands.keys():
            self._currcom['command'] = self.__allowed_commands[data]
        elif data in self.__allowed_commands.values():
            self._currcom['command'] = data
        else: raise ValueError('Unknown command given: %s' % str(data))
    def _d_command(self):
        del self._currcom['command']
    command = property(_g_command, _s_command, _d_command, 'main command to worker')
    
    ##### method property
    def _g_method(self):
        return self._currcom['method']
    def _s_method(self, data):
        if data in self.__allowed_methods.keys():
            self._currcom['method'] = self.__allowed_methods[data]
        elif data in self.__allowed_methods.values():
            self._currcom['method'] = data
        else: raise ValueError('Unknown method given: %s' % str(data))
    def _d_method(self):
        del self._currcom['method']
    method = property(_g_method, _s_method, _d_method, 'main command method to worker')
    
    ##### url property
    def _g_url(self):
        return self._currcom['url']
    def _s_url(self, data):
        if type(data) != str: raise TypeError('Given data is not a string: %s' % str(data))
        self._currcom['url'] = data
    def _d_url(self):
        del self._currcom['url']
    url = property(_g_url, _s_url, _d_url, 'main command method to worker')
    
    ##### js_codes property
    def _g_js_codes(self):
        return self._currcom['js_codes']
    def _s_js_codes(self, data):
        if type(data) != list: raise TypeError('Given data is not a list!')
        for x in data:
            if type(x) != str: raise TypeError('Given data is not a string: %s' % str(x) )
        self._currcom['js_codes'] = data
    def _d_js_codes(self):
        del self._currcom['js_codes']
    js_codes = property(_g_js_codes, _s_js_codes, _d_js_codes, 'js codes list to inject in page')
    
    ##### com_data property
    def _g_com_data(self):
        return self._currcom['com_data']
    def _s_com_data(self, data):
        if type(data) != dict: raise TypeError('Given data is not a dict: %s' % str(data))
        self._currcom['com_data'] = data
    def _d_com_data(self):
        del self._currcom['com_data']
    com_data = property(_g_com_data, _s_com_data, _d_com_data, 'dict with different data to mainproc')
###################################################################
class Drinkins:
    __command = 0
    __ret_data = ''
    __ret_head = ''
    __status = 0
    def __init__(self, command=0, status=0, ret_head='', ret_data=''):
        self.configure(command, status, ret_head, ret_data)
    
    def configure(self, command, status, ret_head='', ret_data=''):
        '''
        for multiple attribute setting
        '''
        self._s_command(command)
        self._s_status(status)
        self._s_ret_head(ret_head)
        self._s_ret_data(ret_data)
        return self
        
    
    ##### command property
    def _g_command(self):
        return self.__command
    def _s_command(self, data):
        if type(data) != int: raise TypeError('data is int not %s' % type(data))
        self.__command = data
    def _d_command(self):
        del self.__command
    command = property(_g_command, _s_command, _d_command, 'copy command code from Comandor instance')
    
    ##### data from processed command property
    def _g_ret_data(self):
        return self.__ret_data
    def _s_ret_data(self, data):
        self.__ret_data = data
    def _d_ret_data(self):
        del self.__ret_data
    ret_data = property(_g_ret_data, _s_ret_data, _d_ret_data, 'answer data for Process parsing command')
    
    ##### data from processed command property
    def _g_ret_head(self):
        return self.__ret_head
    def _s_ret_head(self, data):
        self.__ret_head = data
    def _d_ret_head(self):
        del self.__ret_head
    ret_head = property(_g_ret_head, _s_ret_head, _d_ret_head, 'answer head data for Process parsing command')
    
    ##### command status property
    def _g_status(self):
        return self.__status
    def _s_status(self, data):
        if type(data) != int: raise TypeError('data is not int: %s' % type(data))
        self.__status = data
    def _d_status(self):
        del self.__status
    status = property(_g_status, _s_status, _d_status, 'status of processing command')

вот так делаю в дочернем процессе:

 #... some skipped code here
def run(self, master, slave, superloop, comm):
        ncom = Comandor(command=4, com_data = {'fuck':'self._get_procinfo()'})
        slave.put(ncom) # slave - multiprocessing.Queue for store commands from multiprocessing.Process by Comandor instance
        print('PROCESSWORKER: slave.put(self.comm)', ncom, ncom.command, ncom.com_data)
        answer = master.get() # master - multiprocessing.Queue for store answers from main process by Drinkins instance
        print('PROCESSWORKER: answer = master.get()', answer, answer.status)

и вот так в главном процессе

 #... some skipped code here
task = self.p_algo.slave.get()
print('MAINWORKER: task = self.p_algo.slave.get()', task, task.command, task.com_data)
#... some skipped code here
self.answer.configure(command=task.command, status=10) # set Drinkins values
print('MAINWORKER: self.p_algo.master.put(self.answer)', self.answer)
self.p_algo.master.put(self.answer)

при этом в консоль печатает :

PROCESSWORKER: slave.put(self.comm) <communication.Comandor object at 0x0000000002B08198> 4 {'fuck': 'self._get_procinfo()'}
MAINWORKER: task = self.p_algo.slave.get() <communication.Comandor object at 0x0000000006538438> 0 {}
MAINWORKER: self.p_algo.master.put(self.answer) <plugins.communicator.communication.Drinkins object at 0x0000000004332588>
PROCESSWORKER: answer = master.get() <plugins.communicator.communication.Drinkins object at 0x00000000045C41D0> 10

из напечатанного в консоли видно что очередность обмена сообщениями работает, но инстанс командора улетал из дочернего процесса с выставленными значениями а прилетел в главный с дефолтными, при этом если в дочернем отправлять инстанс Drinkins или Comandor.as_dict() то все прилетает норм.

Народ, выручайте, уже всю голову изломал но не пойму в чем дело.
@cckyi_boxxx
Спасибо всем кто пытался разобраться, выяснил в чем была проблема. Оказывается при использовании property() совместно с мультипроцессингом нельзя привязывать их к элементам словаря, сейчас создал для каждого вычисляемого свойства свою отдельную переменную, так-же как это сделано в классе Drinkins и все заработало как часы.
This is a "lo-fi" version of our main content. To view the full version with more information, formatting and images, please click here.
Powered by DjangoBB