multiprocesamiento: ¿Cómo comparto un dictado entre múltiples procesos?

113

Un programa que crea varios procesos que funcionan en una cola que se puede unir Q, y que eventualmente puede manipular un diccionario global Dpara almacenar resultados. (por lo que cada proceso hijo puede usar Dpara almacenar su resultado y también ver qué resultados están produciendo los otros procesos secundarios)

Si imprimo el diccionario D en un proceso hijo, veo las modificaciones que se han realizado en él (es decir, en D). Pero después de que el proceso principal se une a Q, si imprimo D, ¡es un dict vacío!

Entiendo que es un problema de sincronización / bloqueo. ¿Alguien puede decirme qué está sucediendo aquí y cómo puedo sincronizar el acceso a D?

dop
fuente
1
Esto no funciona como se esperaba al menos en Python 3.7.2 usando osx 10.14.4 Dict no está sincronizado y su contenido es reescrito por otros procesos. Sin embargo, <code> multiprocessing.Manager (). List () </code> funciona como se esperaba.
Andrew Druchenko

Respuestas:

162

Una respuesta general implica usar un Managerobjeto. Adaptado de los documentos:

from multiprocessing import Process, Manager

def f(d):
    d[1] += '1'
    d['2'] += 2

if __name__ == '__main__':
    manager = Manager()

    d = manager.dict()
    d[1] = '1'
    d['2'] = 2

    p1 = Process(target=f, args=(d,))
    p2 = Process(target=f, args=(d,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()

    print d

Salida:

$ python mul.py 
{1: '111', '2': 6}
remitente
fuente
4
Gracias senderle. De hecho, D = multiprocessing.Manager (). Dict () resuelve mi problema. Estaba usando D = dict ().
dop
3
@LorenzoBelli, si estás preguntando si el acceso al administrador está sincronizado, creo que la respuesta es sí. multiprocessing.Manager()devuelve una instancia deSyncManager , cuyo nombre sugiere tanto!
remitente
@senderle Quiero compartir un gran estado aleatorio de un proceso principal con un proceso secundario. Intenté usar Managerpero aún no tuve suerte. ¿Podría echar un vistazo a mi pregunta aquí y ver si puede ofrecer una solución? Aún puedo obtener diferentes números aleatorios si lo hago np.random.seed(None)cada vez que genero un número aleatorio, pero esto no me permite usar el estado aleatorio del proceso principal, que no es lo que quiero. Cualquier ayuda es muy apreciada.
Amir
1
@RadioControlled feliz de escribir una actualización, pero brevemente, aunque no creo que pueda hacer que esto suceda directamente, puede crear fácilmente un nuevo dictado administrado con las mismas claves y valores, y usarlo en lugar del original. ¿Es eso adecuado para su caso?
remitente
1
@senderle, eso es lo que terminé haciendo. Entonces la respuesta sería que tendrías que hacer precisamente eso.
Radio controlado
25

multiprocesamiento no es como enhebrar. Cada proceso hijo obtendrá una copia de la memoria del proceso principal. Generalmente, el estado se comparte a través de comunicación (tuberías / enchufes), señales o memoria compartida.

El multiprocesamiento hace que algunas abstracciones estén disponibles para su caso de uso: estado compartido que se trata como local mediante el uso de proxies o memoria compartida: http://docs.python.org/library/multiprocessing.html#sharing-state-between-processes

Secciones relevantes:

Jeremy Brown
fuente
1
Muchas gracias. Me llevó a la solución / a: multiprocessing.Manager (). Dict ().
dop
¿Alguien puede explicar qué significa la afirmación "Cada proceso hijo obtendrá una copia de la memoria del proceso principal"?
Itsme2003
@ Itsme2003 de forma predeterminada, un proceso generado no tiene acceso a la memoria del proceso principal (esta es una de las diferencias clave de los subprocesos). Entonces, cuando un proceso necesita un objeto del proceso padre, tiene que crear una copia del mismo (en lugar de obtener una referencia al objeto real). La respuesta anterior explica cómo compartir objetos entre procesos.
Niklas Mertsch
Porque esto a menudo se confunde: siempre que no modifique el objeto, al menos en la configuración habitual de Linux, el objeto solo se almacenará una vez en la memoria. Se copiará tan pronto como se modifique. Esto puede ser muy importante si necesita ahorrar memoria y no modifica el objeto.
Radio controlado
16

Me gustaría compartir mi propio trabajo que es más rápido que el dictado de Manager y es más simple y más estable que la biblioteca pyshmht que usa toneladas de memoria y no funciona para Mac OS. Aunque mi dictado solo funciona para cadenas simples y actualmente es inmutable. Uso la implementación de sondeo lineal y almaceno pares de claves y valores en un bloque de memoria separado después de la tabla.

from mmap import mmap
import struct
from timeit import default_timer
from multiprocessing import Manager
from pyshmht import HashTable


class shared_immutable_dict:
    def __init__(self, a):
        self.hs = 1 << (len(a) * 3).bit_length()
        kvp = self.hs * 4
        ht = [0xffffffff] * self.hs
        kvl = []
        for k, v in a.iteritems():
            h = self.hash(k)
            while ht[h] != 0xffffffff:
                h = (h + 1) & (self.hs - 1)
            ht[h] = kvp
            kvp += self.kvlen(k) + self.kvlen(v)
            kvl.append(k)
            kvl.append(v)

        self.m = mmap(-1, kvp)
        for p in ht:
            self.m.write(uint_format.pack(p))
        for x in kvl:
            if len(x) <= 0x7f:
                self.m.write_byte(chr(len(x)))
            else:
                self.m.write(uint_format.pack(0x80000000 + len(x)))
            self.m.write(x)

    def hash(self, k):
        h = hash(k)
        h = (h + (h >> 3) + (h >> 13) + (h >> 23)) * 1749375391 & (self.hs - 1)
        return h

    def get(self, k, d=None):
        h = self.hash(k)
        while True:
            x = uint_format.unpack(self.m[h * 4:h * 4 + 4])[0]
            if x == 0xffffffff:
                return d
            self.m.seek(x)
            if k == self.read_kv():
                return self.read_kv()
            h = (h + 1) & (self.hs - 1)

    def read_kv(self):
        sz = ord(self.m.read_byte())
        if sz & 0x80:
            sz = uint_format.unpack(chr(sz) + self.m.read(3))[0] - 0x80000000
        return self.m.read(sz)

    def kvlen(self, k):
        return len(k) + (1 if len(k) <= 0x7f else 4)

    def __contains__(self, k):
        return self.get(k, None) is not None

    def close(self):
        self.m.close()

uint_format = struct.Struct('>I')


def uget(a, k, d=None):
    return to_unicode(a.get(to_str(k), d))


def uin(a, k):
    return to_str(k) in a


def to_unicode(s):
    return s.decode('utf-8') if isinstance(s, str) else s


def to_str(s):
    return s.encode('utf-8') if isinstance(s, unicode) else s


def mmap_test():
    n = 1000000
    d = shared_immutable_dict({str(i * 2): '1' for i in xrange(n)})
    start_time = default_timer()
    for i in xrange(n):
        if bool(d.get(str(i))) != (i % 2 == 0):
            raise Exception(i)
    print 'mmap speed: %d gets per sec' % (n / (default_timer() - start_time))


def manager_test():
    n = 100000
    d = Manager().dict({str(i * 2): '1' for i in xrange(n)})
    start_time = default_timer()
    for i in xrange(n):
        if bool(d.get(str(i))) != (i % 2 == 0):
            raise Exception(i)
    print 'manager speed: %d gets per sec' % (n / (default_timer() - start_time))


def shm_test():
    n = 1000000
    d = HashTable('tmp', n)
    d.update({str(i * 2): '1' for i in xrange(n)})
    start_time = default_timer()
    for i in xrange(n):
        if bool(d.get(str(i))) != (i % 2 == 0):
            raise Exception(i)
    print 'shm speed: %d gets per sec' % (n / (default_timer() - start_time))


if __name__ == '__main__':
    mmap_test()
    manager_test()
    shm_test()

En mi computadora portátil, los resultados de rendimiento son:

mmap speed: 247288 gets per sec
manager speed: 33792 gets per sec
shm speed: 691332 gets per sec

ejemplo de uso simple:

ht = shared_immutable_dict({'a': '1', 'b': '2'})
print ht.get('a')
alyaxey
fuente
14
Github? ¿Documentación? ¿cómo podemos utilizar esta herramienta?
Pavlos Panteliadis
10

Además de @ senderle's here, algunos también podrían preguntarse cómo usar la funcionalidad de multiprocessing.Pool .

Lo bueno es que hay un .Pool()método para la managerinstancia que imita toda la API familiar del nivel superior multiprocessing.

from itertools import repeat
import multiprocessing as mp
import os
import pprint

def f(d: dict) -> None:
    pid = os.getpid()
    d[pid] = "Hi, I was written by process %d" % pid

if __name__ == '__main__':
    with mp.Manager() as manager:
        d = manager.dict()
        with manager.Pool() as pool:
            pool.map(f, repeat(d, 10))
        # `d` is a DictProxy object that can be converted to dict
        pprint.pprint(dict(d))

Salida:

$ python3 mul.py 
{22562: 'Hi, I was written by process 22562',
 22563: 'Hi, I was written by process 22563',
 22564: 'Hi, I was written by process 22564',
 22565: 'Hi, I was written by process 22565',
 22566: 'Hi, I was written by process 22566',
 22567: 'Hi, I was written by process 22567',
 22568: 'Hi, I was written by process 22568',
 22569: 'Hi, I was written by process 22569',
 22570: 'Hi, I was written by process 22570',
 22571: 'Hi, I was written by process 22571'}

Este es un ejemplo ligeramente diferente en el que cada proceso simplemente registra su ID de proceso en el DictProxyobjeto global d.

Brad Solomon
fuente
3

Tal vez puedas probar pyshmht , que comparte la extensión de la tabla hash basada en memoria para Python.

aviso

  1. No está completamente probado, solo para su referencia.

  2. Actualmente carece de mecanismos lock / sem para multiprocesamiento.

felix021
fuente