Обновить объект внутри мультипроцессинга.Manager.dict ()

Question

Мне было интересно, как я могу обновить объект, назначенный в качестве значения общего словаря, между различными процессами. У меня есть следующий класс:

class Task:

    STATUS_PROCESSING = 0
    STATUS_EXECUTING = 1
    STATUS_QUEUED = 2
    STATUS_TERMINATED = 3
    STATUS_HALTED = 4
    STATUS_STOPPED = 5

    def __init__(self, id: str, uuid: str, options: dict):
        self.id = id
        self.uuid = uuid
        self.options = options
        self.state = 0

    # Some properties...

    def execute(self):
        """ Executes the task
        """
        # Set self status to Executing
        self.state = Task.STATUS_EXECUTING

        print('Executing...')

        self.state = Task.STATUS_TERMINATED

Он просто создает новую задачу с заданным идентификатором и выполняет основной метод при вызове execute(). У меня есть другой класс со staticmethods, который используется для добавления новой пары (id, task) к dict и чтения dict, выполняющего все его задачи до остановки основной программы:

class DummyList:

    @staticmethod
    def submit_task(d: dict, uuid: str, options: dict):
        """ Submit a new task
        """
        # If invalid UUID
        if not Task.is_valid_uuid(uuid):
            return False

        # If more than 20 tasks
        if len(d) > 19:
            return False

        # Create random ID (simplified for question)
        r_id = str(random.randint(1, 2000000))
        if r_id in d:
            return False

        # Add task to the dictionary
        d[r_id] = Task(r_id, uuid, options)

        # Set status to queue
        d[r_id].state = Task.STATUS_QUEUED

        # Return the created ID
        return r_id

    @staticmethod
    def execute_forever(d):
        try:
            while True:
                for i in d.values():
                    print(i.state)
                    i.execute()
                time.sleep(5)
        except KeyboardInterrupt:
            pass

Дело в том, что DummyList.execute_forever() будет вызываться из другого процесса, а основной будет выполнять функцию submit_task(...) для добавления новых задач. Вот так:

        # Create a shared dict
        m = multiprocessing.Manager()
        shared_d = m.dict()

        # Start the Task shared list execution in another process
        p = multiprocessing.Process(target=DummyList.execute_forever, args=(shared_d,))
        # Set the process to exit when the main halts
        p.daemon = True
        p.start()

        ........


       # From another place
       # The message variable is not important
       DummyList.submit_task(shared_d, message['proc'], message['options'])

Работает! Задача создана, присвоена словарю и выполнена, но следующие строки (которые видны в приведенном выше коде) не выполняются должным образом:

self.state = Task.STATUS_EXECUTING
self.state = Task.STATUS_TERMINATED
d[r_id].state = Task.STATUS_QUEUED

Если мы попытаемся написать ìf shared_d[<some_id>].state == 0 во всем коде это всегда будет True, потому что свойство не обновляется

Я полагаю, это потому, что общий словарь не обновляется при изменении свойств объекта, возможно, потому что словарь он понимает, что должен обновляться только при вызове его методов getitem или setitem . Знаете ли вы, есть ли способ изменить это поведение?

Большое спасибо!

Answer 1

Я наконец нашел решение. Объекты внутри словаря не обновлялись, пока не были вызваны методы __getitem__ или __setitem__ из прокси-словаря. Вот почему я изменил следующие строки:

Задача

Метод execute() заканчивается на return self. self.state должен быть изменен во время выполнения.

TaskManager

Метод изменен на:

@staticmethod
    def execute_forever(d):
        """ Infinite loop reading the queued tasks and executing all of them.
        """
        try:
            while True:
                # Notice the loop using the keys
                for i in d.keys():
                    # Execute and re-assign item
                    d[i] = d[i].execute()
                time.sleep(5)
        except KeyboardInterrupt:
            pass