pildorasinformaticas - python 17

Paralelismo anidado en Python

Question

Estoy probando la programación multiprocesador con Python. Tome un algoritmo de dividir y conquistar como Fibonacci por ejemplo. El flujo de ejecución del programa se ramificaría como un árbol y se ejecutaría en paralelo. En otras palabras, tenemos un ejemplo de paralelismo anidado .

Desde Java, he usado un patrón de subprocesos para administrar recursos, ya que el programa podría extenderse muy rápidamente y crear demasiados subprocesos de corta duración. Un solo grupo de subprocesos estático (compartido) se puede crear una instancia a través de ExecutorService .

Yo esperaría lo mismo para Pool , pero parece que el objeto Pool no debe compartirse globalmente . Por ejemplo, compartir el grupo utilizando multiprocessing.Manager.Namespace() llevará al error.

Los objetos del pool no se pueden pasar entre procesos o decapados.

Tengo una pregunta de 2 partes:

1. Que me estoy perdiendo aqui; ¿Por qué no se debería compartir un Pool entre procesos?
2. ¿Cuál es un patrón para implementar el paralelismo anidado en Python? Si es posible, mantener una estructura recursiva y no cambiarla por iteración.

from concurrent.futures import ThreadPoolExecutor

def fibonacci(n):
    if n < 2:
        return n
    a = pool.submit(fibonacci, n - 1)
    b = pool.submit(fibonacci, n - 2)
    return a.result() + b.result()

def main():
    global pool

    N = int(10)
    with ThreadPoolExecutor(2**N) as pool:
        print(fibonacci(N))

main()

Java

public class FibTask implements Callable<Integer> {

    public static ExecutorService pool = Executors.newCachedThreadPool();
    int arg;

    public FibTask(int n) {
        this.arg= n;
    }

    @Override
    public Integer call() throws Exception {
        if (this.arg > 2) { 
            Future<Integer> left = pool.submit(new FibTask(arg - 1));
            Future<Integer> right = pool.submit(new FibTask(arg - 2));
            return left.get() + right.get();
        } else {
            return 1;
        }

    } 

  public static void main(String[] args) throws Exception {
      Integer n = 14;
      Callable<Integer> task = new FibTask(n);
      Future<Integer> result =FibTask.pool.submit(task); 
      System.out.println(Integer.toString(result.get()));
      FibTask.pool.shutdown();            
  }    

}

No estoy seguro si importa aquí, pero estoy ignorando la diferencia entre "proceso" y "hilo"; Para mí, ambos significan "procesador virtualizado". Mi entendimiento es que el propósito de un Pool es compartir un "pool" o recursos. Las tareas en ejecución pueden hacer una solicitud al Pool. A medida que las tareas paralelas se completan en otros subprocesos, dichos subprocesos se pueden reclamar y asignar a nuevas tareas. No tiene sentido para mí no permitir compartir el grupo, por lo que cada hilo debe crear una instancia de su propio grupo nuevo, ya que eso parece anular el propósito de un grupo de hilos.

Answer 1

1) ¿Qué me estoy perdiendo aquí? ¿Por qué no se debería compartir un Pool entre procesos?

No todos los objetos / instancias son seleccionables / serializables, en este caso, el grupo usa threading.lock que no es seleccionable:

>>> import threading, pickle
>>> pickle.dumps(threading.Lock())
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
[...]
  File "/Users/rafael/dev/venvs/general/bin/../lib/python2.7/copy_reg.py", line 70, in _reduce_ex
    raise TypeError, "can''t pickle %s objects" % base.__name__
TypeError: can''t pickle lock objects

o mejor:

>>> import threading, pickle
>>> from concurrent.futures import ThreadPoolExecutor
>>> pickle.dumps(ThreadPoolExecutor(1))
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/usr/local/Cellar/python/2.7.3/Frameworks/Python.framework/Versions/2.7/lib/python2.7/pickle.py", line 1374, in dumps
    Pickler(file, protocol).dump(obj)
  File 
[...]
"/usr/local/Cellar/python/2.7.3/Frameworks/Python.framework/Versions/2.7/lib/python2.7/pickle.py", line 306, in save
        rv = reduce(self.proto)
      File "/Users/rafael/dev/venvs/general/bin/../lib/python2.7/copy_reg.py", line 70, in _reduce_ex
        raise TypeError, "can''t pickle %s objects" % base.__name__
    TypeError: can''t pickle lock objects

Si lo piensas bien, tiene sentido, un bloqueo es una primitiva de semáforo administrada por el sistema operativo (ya que Python utiliza subprocesos nativos). Ser capaz de decapar y guardar ese estado de objeto dentro del tiempo de ejecución de python realmente no lograría nada significativo, ya que su verdadero estado está siendo mantenido por el sistema operativo.

2) ¿Cuál es un patrón para implementar el paralelismo anidado en Python? Si es posible, mantener una estructura recursiva y no cambiarla por iteración.

Ahora, por el prestigio, todo lo que mencioné anteriormente no se aplica realmente a su ejemplo, ya que está utilizando subprocesos (ThreadPoolExecutor) y no procesos (ProcessPoolExecutor), por lo que no es necesario que se compartan los datos.

Su ejemplo de Java parece ser más eficiente ya que el grupo de subprocesos que está utilizando (CachedThreadPool) está creando nuevos subprocesos según sea necesario, mientras que las implementaciones del ejecutor de Python están limitadas y requieren un número máximo de subprocesos explícito (max_workers). Hay un poco de diferencias de sintaxis entre los idiomas que también parecen estar eliminándote (las instancias estáticas en Python son esencialmente algo que no tienen un alcance explícito) pero esencialmente ambos ejemplos crearían exactamente el mismo número de subprocesos para poder ejecutar. Por ejemplo, aquí hay un ejemplo que utiliza una implementación bastante ingenua de CachedThreadPoolExecutor en python:

from concurrent.futures import ThreadPoolExecutor

class CachedThreadPoolExecutor(ThreadPoolExecutor):
    def __init__(self):
        super(CachedThreadPoolExecutor, self).__init__(max_workers=1)

    def submit(self, fn, *args, **extra):
        if self._work_queue.qsize() > 0:
            print(''increasing pool size from %d to %d'' % (self._max_workers, self._max_workers+1))
            self._max_workers +=1

        return super(CachedThreadPoolExecutor, self).submit(fn, *args, **extra)

pool = CachedThreadPoolExecutor()

def fibonacci(n):
    print n
    if n < 2:
        return n
    a = pool.submit(fibonacci, n - 1)
    b = pool.submit(fibonacci, n - 2)
    return a.result() + b.result()

print(fibonacci(10))

La optimización del rendimiento:

Recomiendo encarecidamente buscar en gevent ya que le dará una alta concurrencia sin la sobrecarga de subprocesos. Este no es siempre el caso, pero su código es en realidad el elemento secundario para el uso de gevent Aquí hay un ejemplo:

import gevent

def fibonacci(n):
    print n
    if n < 2:
        return n
    a = gevent.spawn(fibonacci, n - 1)
    b = gevent.spawn(fibonacci, n - 2)
    return a.get()  + b.get()

print(fibonacci(10))

Completamente no científico, pero en mi computadora el código anterior se ejecuta 9 veces más rápido que su equivalente en subprocesos.

Espero que esto ayude.

Answer 2

1. ¿Qué me estoy perdiendo aquí? ¿Por qué no se debería compartir un Pool entre procesos?

Por lo general, no puede compartir subprocesos del sistema operativo entre procesos, independientemente del idioma.

Puede organizar compartir el acceso al administrador de grupo con los procesos de trabajo, pero probablemente no sea una buena solución para cualquier problema; vea abajo.

2. ¿Cuál es un patrón para implementar el paralelismo anidado en Python? Si es posible, mantener una estructura recursiva y no cambiarla por iteración.

Esto depende mucho de sus datos.

En CPython, la respuesta general es usar una estructura de datos que implemente operaciones paralelas eficientes. Un buen ejemplo de esto son los tipos de arreglos optimizados de NumPy : here hay un ejemplo de cómo usarlos para dividir una gran operación de arreglos en múltiples núcleos de procesadores.

Sin embargo, la función de Fibonacci implementada mediante el uso de la recursión de bloqueo es un ajuste particularmente pesimista para cualquier enfoque basado en el grupo de trabajadores: fib (N) pasará gran parte de su tiempo solo atando a N trabajadores haciendo nada más que esperar a otros trabajadores. Hay muchas otras formas de abordar la función de Fibonacci específicamente (por ejemplo, usar CPS para eliminar el bloqueo y llenar un número constante de trabajadores), pero probablemente sea mejor decidir su estrategia en función de los problemas reales que resolverá, en lugar de ejemplos. Me gusta esto.