Как синхронизировать python dict с многопроцессорной обработкой

Ответ 1

Введение

Кажется, что есть много предложений для кресла и рабочих примеров. Ни один из приведенных здесь ответов даже не предполагает использование многопроцессорности, и это довольно неприятно и тревожно. Будучи любителями python, мы должны поддерживать наши встроенные библиотеки, и, хотя параллельная обработка и синхронизация никогда не являются тривиальным вопросом, я считаю, что это может быть тривиально с правильным дизайном. Это становится чрезвычайно важным в современных многоядерных архитектурах и не может быть достаточно напряженным! Тем не менее, я далек от удовлетворения многопроцессорной библиотеки, поскольку она все еще находится на стадии зачатия с довольно большим количеством ошибок, ошибок и привязки к функциональному программированию (которое я ненавижу). В настоящее время я по-прежнему предпочитаю модуль Pyro (который намного опережает свое время) по многопроцессорной обработке из-за многопроцессорности серьезных ограничений в том, созданных объектов во время работы сервера. Класс-метод "register" объектов-менеджеров будет фактически зарегистрировать объект только до того, как запущен менеджер (или его сервер). Достаточно болтовни, больше кода:

Server.py

from multiprocessing.managers import SyncManager


class MyManager(SyncManager):
    pass


syncdict = {}
def get_dict():
    return syncdict

if __name__ == "__main__":
    MyManager.register("syncdict", get_dict)
    manager = MyManager(("127.0.0.1", 5000), authkey="password")
    manager.start()
    raw_input("Press any key to kill server".center(50, "-"))
    manager.shutdown()

В приведенном выше примере кода Server.py использует многопроцессорный SyncManager, который может поставлять синхронизированные общие объекты. Этот код не будет работать в интерпретаторе, потому что библиотека многопроцессорности довольно трогательна в отношении того, как найти "вызываемый" для каждого зарегистрированного объекта. Запуск Server.py запустит настроенный SyncManager, который будет использовать словарь syncdict для использования нескольких процессов и может быть подключен к клиентам либо на одном компьютере, либо если выполняется по IP-адресу, отличному от loopback, другим машинам. В этом случае сервер запускается по шлейфу (127.0.0.1) на порту 5000. Использование параметра authkey использует защищенные соединения при манипулировании синхронизацией. При нажатии любой клавиши менеджер отключается.

Client.py

from multiprocessing.managers import SyncManager
import sys, time

class MyManager(SyncManager):
    pass

MyManager.register("syncdict")

if __name__ == "__main__":
    manager = MyManager(("127.0.0.1", 5000), authkey="password")
    manager.connect()
    syncdict = manager.syncdict()

    print "dict = %s" % (dir(syncdict))
    key = raw_input("Enter key to update: ")
    inc = float(raw_input("Enter increment: "))
    sleep = float(raw_input("Enter sleep time (sec): "))

    try:
         #if the key doesn't exist create it
         if not syncdict.has_key(key):
             syncdict.update([(key, 0)])
         #increment key value every sleep seconds
         #then print syncdict
         while True:
              syncdict.update([(key, syncdict.get(key) + inc)])
              time.sleep(sleep)
              print "%s" % (syncdict)
    except KeyboardInterrupt:
         print "Killed client"

Клиент также должен создать настраиваемый SyncManager, зарегистрировав "syncdict", на этот раз, не передав вызываемому, чтобы получить общий dict. Затем он использует настроенный SycnManager для подключения с использованием IP-адреса loopback (127.0.0.1) на порту 5000 и авторизацию, устанавливающую безопасное соединение с менеджером, запущенным на Server.py. Он извлекает общий синдикат dict, вызывая зарегистрированный вызываемый в менеджере. Он запрашивает у пользователя следующее:

• Ключ в синклюке для работы
• Сумма, увеличивающая значение, доступное ключу в каждом цикле
• Время ожидания для каждого цикла в секундах

Затем клиент проверяет, существует ли ключ. Если это не так, он создает ключ в синцитете. Затем клиент вводит "бесконечный" цикл, в котором он обновляет значение ключа с помощью инкремента, присваивает указанную сумму и печатает синкатдик только для повторения этого процесса до тех пор, пока не произойдет KeyboardInterrupt (Ctrl + C).

Раздражающие проблемы

• Способы регистрации менеджера ДОЛЖНЫ вызываться до запуска менеджера, иначе вы получите исключения, даже если вызов dir в диспетчере покажет, что он действительно имеет зарегистрированный метод.
• Все манипуляции с dict должны выполняться с помощью методов, а не диктовых назначений (syncdict [ "blast" ] = 2 будет терпеть неудачу из-за того, как многопроцессор делится пользовательскими объектами)
• Использование метода синтаксиса SyncManager облегчит неприятную проблему №2, за исключением того, что раздражающая проблема # 1 предотвращает регистрацию и совместное использование прокси, возвращаемого SyncManager.dict(). (SyncManager.dict() может быть вызван только после запуска менеджера, и регистрация будет работать только до того, как менеджер запустится, поэтому SyncManager.dict() полезен только при выполнении функционального программирования и передаче прокси-сервера процессам в качестве аргумента, такого как doc примеры)
• Сервер И клиент должны регистрироваться, хотя интуитивно казалось бы, что клиент сможет просто понять его после подключения к менеджеру (добавьте это в свои разработчики многопроцессорности списка пожеланий)

Закрытие

Надеюсь, вам понравился этот довольно тщательный и немного отнимающий много времени ответ, как я. У меня возникло множество проблем, возникающих у меня в голове, почему я так много борется с модулем многопроцессорности, где Пиро делает его легким и теперь благодаря этому ответу я ударил ноготь по голове. Я надеюсь, что это полезно для сообщества python о том, как улучшить модуль многопроцессорности, поскольку я считаю, что он имеет большие перспективы, но в зачаточном состоянии отстает от того, что возможно. Несмотря на раздражающие проблемы, я думаю, что это все еще довольно жизнеспособная альтернатива и довольно проста. Вы также можете использовать SyncManager.dict() и передать его процессам в качестве аргумента, как показывают документы, и, вероятно, это будет еще более простое решение в зависимости от ваших требований, которое мне кажется неестественным для меня.

Ответ 2

Я бы посвятил отдельный процесс поддержанию "общего дикта": просто используйте, например. xmlrpclib, чтобы сделать этот крошечный код доступным для других процессов, выставляя через xmlrpclib, например. функция, принимающая key, increment для выполнения приращения, и один, принимающий только key и возвращающий значение, с семантическими деталями (есть ли значение по умолчанию для отсутствующих ключей и т.д. и т.д.) в зависимости от потребностей вашего приложения.

Затем вы можете использовать любой подход, который вам нравится реализовать совместно выделенный процесс: полностью от однопоточного сервера с простым dict в памяти, до простой sqlite DB и т.д. и т.д. Я предлагаю вам начать с кодом "так же просто, как вы можете уйти" (в зависимости от того, нужен ли вам постоянный общий dict, или вам не требуется постоянство), затем измерьте и оптимизируйте как можно, и если нужно.

Ответ 3

В ответ на соответствующее решение проблемы одновременной записи. Я сделал очень быстрое исследование и обнаружил, что в этой статье предлагает решение для блокировки/семафора. (http://effbot.org/zone/thread-synchronization.htm)

В то время как пример не является спецификой словаря, я уверен, что вы можете закодировать объект-оболочку на основе класса, чтобы помочь вам работать со словарями на основе этой идеи.

Если бы у меня было требование реализовать что-то вроде этого в потокобезопасном режиме, я бы, вероятно, использовал решение Python Semaphore. (Предполагая, что мой предыдущий метод слияния не будет работать.) Я считаю, что семафоры обычно замедляют эффективность потоков из-за их блокирующего характера.

С сайта:

Семафор - это более продвинутый механизм блокировки. Семафор имеет внутренний счетчик, а не флаг блокировки, и он блокируется только в том случае, если более чем определенное количество потоков пыталось удерживать семафор. В зависимости от того, как инициализируется семафор, это позволяет нескольким потокам одновременно обращаться к одной и той же секции кода.

semaphore = threading.BoundedSemaphore()
semaphore.acquire() # decrements the counter
... access the shared resource; work with dictionary, add item or whatever.
semaphore.release() # increments the counter

Ответ 4

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

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