Módulo de multiprocesamiento para actualizar un diccionario compartido en Python

Debido a que desea aumentar los recuentos (no solo crear nuevos pares clave / valor), el diccionario no es seguro para usar con hilos a menos que adquiera un semáforo alrededor de cada actualización y lo libere después, por lo que no creo que obtenga una velocidad general ganancia, de hecho puede ser más lento.

Si va a enhebrar esto, probablemente sea mejor que cada hilo actualice su propio diccionario y luego combine los resultados a medida que finaliza cada hilo, de esa manera no hay duda sobre la seguridad del hilo. Sin embargo, debido a que es probable que esté vinculado a la CPU, debe usar multiprocesamiento en lugar de subprocesos; el multiprocesamiento puede hacer uso de todos los núcleos de su CPU.

Además, si usa colecciones.Counter, hará el recuento por usted, y admite la fusión, y tiene un método útil most_common (n) para devolver las claves con n recuentos más altos.

Puede seguir un enfoque MapReduce.

from collections import Counter
from multiprocessing import Pool

NUM_PROCESSES = 8

y = [(1,2),(2,3),(1,2),(5,6)] * 10

## http://stackoverflow.com/questions/312443/how-do-you-split-a-list-into-evenly-sized-chunks-in-python
def chunks(l, n):
    """Yield successive n-sized chunks from l."""
    for i in xrange(0, len(l), n):
        yield l[i:i+n]

## map
partial_counters = Pool(NUM_PROCESSES).map(Counter, chunks(y, NUM_PROCESSES))

## reduce
reduced_counter = reduce(Counter.__add__, partial_counters)

## Result is:
## Counter({(1, 2): 20, (5, 6): 10, (2, 3): 10})

La idea es:

1. divide tu lista de entrada en trozos
2. alimentar cada fragmento a un proceso separado que calculará de forma independiente los recuentos
3. fusionó todos los recuentos parciales mediante una operación de reducción.

EDITAR: use chunks(map(frozenset, y), NUM_PROCESSES) para tener en cuenta los pares desordenados.

En primer lugar, en lugar de verificar la membresía de tup en dict.keys en cada iteración, lo cual es una muy mala idea, puede usar collections.defaultdict() para este objetivo, que es más pitónico:

from collections import defaultdict
test_dict = defaultdict(lambda:1)

for tup in y:
    tup=tuple(sorted(tup))
    test_dict[tup]=+1

En segundo lugar, si desea utilizar la simultaneidad, es posible que desee usar multiprocesamiento o multiprocesamiento, pero sobre multiprocesamiento, debido a GIL varios subprocesos no pueden ejecutar un bytecode a la vez, y no puede atravesar sus tuplas de ambos lados como un algoritmo BDS.

Pero para el procesamiento múltiple tendrá otro problema: acceder a una memoria compartida desde cada núcleo y trabajar en él de inmediato para obtener más información, lea esta respuesta https://stackoverflow.com/a/30132673/2867928.

Entonces, ¿cuál sería el truco?

Una forma es dividir su lista en piezas pequeñas y usar subprocesos múltiples para aplicar su función en una parte específica.

Otra forma es usar coroutinas y subrutinas que, como se mencionó en esa respuesta, Greg Ewing tiene un gran demostración de cómo usar el rendimiento de usar corutinas para construir cosas como planificadores o simulaciones de muchos actores.

Editar: Respuesta editada para ser segura para subprocesos

El multiprocessing lo hace fácil.

Simplemente refactorice su código para que el procesamiento se realice en una función:

def process_tuple(tuples):
    count_dict = {}
    for tuple_ in tuples:
        tuple_=tuple(sorted(tuple_))
        if tuple_ in count_dict:
            count_dict[tuple_] += 1
        else:
            count_dict[tuple_] = 1
    return count_dict

Divida la lista de tuplas en un grupo pequeño y luego use {{X0 }} para procesar todos tus grupos.

## http://stackoverflow.com/questions/312443/how-do-you-split-a-list-into-evenly-sized-chunks-in-python
def chunks(l, n):
    """Yield successive n-sized chunks from l."""
    for i in xrange(0, len(l), n):
        yield l[i:i+n]

# cut tuples list into 5 chunks
tuples_groups = chunks(tuples, 5)
pool = Pool(5)
count_dict = {}
# processes chunks in parallel
results = pool.map(process_tuple, tuples_groups)
# collect results
for result in results:
    count_dict.update(result)

multiprocessing.Pool manejará la distribución entre hilos.

Ejemplo completo + puntos de referencia:

import time
import random

start_time = time.time()
tuples = []
x,y = (100000, 10)
for i in range(x):
    tuple_ = []
    for j in range(y):
        tuple_.append(random.randint(0, 9))
    tuples.append(tuple(tuple_))

print("--- %s data generated in %s seconds ---" % (x*y, time.time() - start_time))



def process_tuple(tuples):
    count_dict = {}
    for tuple_ in tuples:
        tuple_=tuple(sorted(tuple_))
        if tuple_ in count_dict:
            count_dict[tuple_] += 1
        else:
            count_dict[tuple_] = 1
    return count_dict

from multiprocessing import Pool

start_time = time.time()

## http://stackoverflow.com/questions/312443/how-do-you-split-a-list-into-evenly-sized-chunks-in-python
def chunks(l, n):
    """Yield successive n-sized chunks from l."""
    for i in xrange(0, len(l), n):
        yield l[i:i+n]

# cut tuples list into 5 chunks
tuples_groups = chunks(tuples, 5)
pool = Pool(5)
count_dict = {}
# processes chunks in parallel
results = pool.map(process_tuple, tuples_groups)
# collect results
for result in results:
    count_dict.update(result)

print("--- Multithread processed in %s seconds ---" % (time.time() - start_time))    



start_time = time.time()
count_dict = {}
for tuple_ in tuples:
    tuple_=tuple(sorted(tuple_))
    if tuple_ in count_dict:
        count_dict[tuple_] += 1
    else:
        count_dict[tuple_] = 1

print("--- Single thread processed in %s seconds ---" % (time.time() - start_time))

--- 10000000 datos generados en 32.7803010941 segundos ---
--- Hilo múltiple procesado en 1.79116892815 segundos ---
--- Hilo único procesado en 2.65010404587 segundos ---