Я использую Python 2.6 и многопроцессорный модуль для многопоточности. Теперь я хотел бы иметь синхронизированный dict (где единственная атомная операция, в которой я действительно нуждаюсь, это оператор + = для значения).
Должен ли я заключить dict с вызовом multiprocessing.sharedctypes.synchronized()? Или это еще один способ пойти?
Кажется, что есть много предложений для кресла и рабочих примеров. Ни один из приведенных здесь ответов даже не предполагает использование многопроцессорности, и это довольно неприятно и тревожно. Будучи любителями python, мы должны поддерживать наши встроенные библиотеки, и, хотя параллельная обработка и синхронизация никогда не являются тривиальным вопросом, я считаю, что это может быть тривиально с правильным дизайном. Это становится чрезвычайно важным в современных многоядерных архитектурах и не может быть достаточно напряженным! Тем не менее, я далек от удовлетворения многопроцессорной библиотеки, поскольку она все еще находится на стадии зачатия с довольно большим количеством ошибок, ошибок и привязки к функциональному программированию (которое я ненавижу). В настоящее время я по-прежнему предпочитаю модуль Pyro (который намного опережает свое время) по многопроцессорной обработке из-за многопроцессорности серьезных ограничений в том, созданных объектов во время работы сервера. Класс-метод "register" объектов-менеджеров будет фактически зарегистрировать объект только до того, как запущен менеджер (или его сервер). Достаточно болтовни, больше кода:
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 использует защищенные соединения при манипулировании синхронизацией. При нажатии любой клавиши менеджер отключается.
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).
Надеюсь, вам понравился этот довольно тщательный и немного отнимающий много времени ответ, как я. У меня возникло множество проблем, возникающих у меня в голове, почему я так много борется с модулем многопроцессорности, где Пиро делает его легким и теперь благодаря этому ответу я ударил ноготь по голове. Я надеюсь, что это полезно для сообщества python о том, как улучшить модуль многопроцессорности, поскольку я считаю, что он имеет большие перспективы, но в зачаточном состоянии отстает от того, что возможно. Несмотря на раздражающие проблемы, я думаю, что это все еще довольно жизнеспособная альтернатива и довольно проста. Вы также можете использовать SyncManager.dict() и передать его процессам в качестве аргумента, как показывают документы, и, вероятно, это будет еще более простое решение в зависимости от ваших требований, которое мне кажется неестественным для меня.
Я бы посвятил отдельный процесс поддержанию "общего дикта": просто используйте, например. xmlrpclib, чтобы сделать этот крошечный код доступным для других процессов, выставляя через xmlrpclib, например. функция, принимающая key, increment для выполнения приращения, и один, принимающий только key и возвращающий значение, с семантическими деталями (есть ли значение по умолчанию для отсутствующих ключей и т.д. и т.д.) в зависимости от потребностей вашего приложения.
Затем вы можете использовать любой подход, который вам нравится реализовать совместно выделенный процесс: полностью от однопоточного сервера с простым dict в памяти, до простой sqlite DB и т.д. и т.д. Я предлагаю вам начать с кодом "так же просто, как вы можете уйти" (в зависимости от того, нужен ли вам постоянный общий dict, или вам не требуется постоянство), затем измерьте и оптимизируйте как можно, и если нужно.
В ответ на соответствующее решение проблемы одновременной записи. Я сделал очень быстрое исследование и обнаружил, что в этой статье предлагает решение для блокировки/семафора. (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
Есть ли причина, по которой словарь должен быть разделен в первую очередь? Могли бы вы, чтобы каждый поток поддерживал свой собственный экземпляр словаря и либо сливался в конце обработки потока, либо периодически использовал обратный вызов для объединения копий отдельных нитевых словарей вместе?
Я точно не знаю, что вы делаете, поэтому держите в себе, что мой письменный план может не работать дословно. То, что я предлагаю, - это скорее идея дизайна на высоком уровне.