<jupyter_start><jupyter_text>**Notas para contenedor de docker:** Comando de docker para ejecución de la nota de forma local:nota: cambiar `` por la ruta de directorio que se desea mapear a `/datos` dentro del contenedor de docker.```docker run --rm -v :/datos --name jupyterlab_numerical -p 8888:8888 -p 8786:8786 -p 8787:8787 -d palmoreck/jupyterlab_numerical:1.1.0```password para jupyterlab: `<PASSWORD>`Detener el contenedor de docker:```docker stop jupyterlab_numerical``` Documentación de la imagen de docker `palmoreck/jupyterlab_numerical:1.1.0` en [liga](https://github.com/palmoreck/dockerfiles/tree/master/jupyterlab/numerical). --- Esta nota utiliza métodos vistos en [1.5.Integracion_numerica](https://github.com/ITAM-DS/analisis-numerico-computo-cientifico/blob/master/temas/I.computo_cientifico/1.5.Integracion_numerica.ipynb)  El módulo de `multiprocessing` Documentación en: [multiprocessing](https://docs.python.org/3.1/library/multiprocessing.html). La implementación más utilizada de Python, [CPython](https://en.wikipedia.org/wiki/CPython) no utiliza múltiples cores por default. De tarea queda leer la discusión de la liga anterior en el apartado *Design* sobre el [Global INterpreter Lock: GIL](https://en.wikipedia.org/wiki/Global_interpreter_lock) y el por qué CPython no soporta ejecución *multithreaded* o *multiprocesses*. El módulo `multiprocessing` nos permite realizar procesamientos basados en procesos o threads para compartir trabajo y datos. Se recomienda usar este módulo para el *shared memory programming* (ver [2.2.Sistemas_de_memoria_compartida](https://github.com/ITAM-DS/analisis-numerico-computo-cientifico/blob/master/temas/II.computo_paralelo/2.2.Sistemas_de_memoria_compartida.ipynb)) y para trabajos que son demandantes de CPU. Para paralelizar trabajos demandantes en I/O no se recomienda su uso. **Otro módulo en Python para procesamiento utilizando los cores de tu máquina es [concurrent.features](https://docs.python.org/3/library/concurrent.futures.html) que provee el comportamiento principal de `multiprocessing`**. Ver [liga](https://stackoverflow.com/questions/38311431/concurrent-futures-processpoolexecutor-vs-multiprocessing-pool-pool?noredirect=1&lq=1) y [liga](https://stackoverflow.com/questions/20776189/concurrent-futures-vs-multiprocessing-in-python-3) para más sobre `concurrent.futures` y `concurrent.futures` vs `multiprocessing`.  Nota sobre el GIL y `multiprocessing` Aunque en Python los threads son nativos del sistema operativo (esto es, no se simulan, son realmente threads del sistema operativo creados en el hardware), están limitados por el *global interpreter lock, GIL*, de modo que un sólo thread interactúe con un objeto Python en un único tiempo. Al usar el módulo `multiprocessing` ejecutamos en paralelo un número de **interpretadores Python** (CPython), cada uno con su propio espacio de memoria privada y su propio GIL que se ejecutan en un instante (y con un thread). **Comentario:** en `multiprocessing` se utilizan subprocesos en lugar de threads.  Ejemplos  Hello world! En `multiprocessing` los procesos son generados al utilizar la clase `Process` para crear objetos y llamar al método `start()`. Ver [Process](https://docs.python.org/3.1/library/multiprocessing.htmlmultiprocessing.Process) para documentación de ésta clase.<jupyter_code>from multiprocessing import Process #importamos clase Process
def f():
    print('hello world! de subproceso')
    
if __name__=='__main__':
    p1 = Process(target=f)
    p2 = Process(target=f)
    p1.start() #start sólo puede ser llamada una vez por objeto Process
    p2.start() 
    p1.join() #el proceso principal espera a que termine p1
    p2.join() #el proceso principal espera a que termine p2
    print('hello world! de proceso')<jupyter_output>hello world! de subproceso
hello world! de subproceso
hello world! de proceso<jupyter_text>**Comentario:** es una buena práctica explícitamente hacer `join`'s para cada objeto process que realizó `start`. Ver [Programming guidelines](https://docs.python.org/3.1/library/multiprocessing.htmlprogramming-guidelines) para buenas prácticas. La clase `Process` recibe la función a ejecutar para cada proceso con el argumento `target` y también tiene `args` para los argumentos de la función:<jupyter_code>def f(s):
    print(s)
    
if __name__=='__main__':
    p1 = Process(target=f, args=('hola',))
    p2 = Process(target=f, args=('mundo',)) 
    p1.start()
    p2.start()
    p1.join()
    p2.join()<jupyter_output>hola
mundo<jupyter_text>**Comentarios:** * De acuerdo a [Programming guidelines](https://docs.python.org/3.1/library/multiprocessing.htmlprogramming-guidelines) se usa `if __name__=='__main__':` para que las personas que tienen Windows puedan ejecutar sin problema el código (para las personas que están en Unix pueden quitar el bloque de `if`). El bloque del `if` ayuda a que los subprocesos importen el módulo `__main__` (por lo que no se ejecuta la sección que está dentro de `if __name__=='__main__':` pues no son programas principales) y continúa la ejecución de las líneas de `start` (cada subproceso ejecuta `f`) y `join`. Si se quita este statement por ejemplo:```def f(s):    print(s)p1 = Process(target=f, args=('hola',))p2 = Process(target=f, args=('mundo',)) p1.start()p2.start()p1.join()p2.join()```el notebook quedará bloqueado pues una celda con el código anterior creará subprocesos que a su vez crearán otros subprocesos, que a su vez crearán otros subprocesos... y así de forma recursiva.* Los argumentos tienen que ser objetos *pickable* o serializados. Ver [what can be pickled and unpickled](https://docs.python.org/3/library/pickle.htmlwhat-can-be-pickled-and-unpickled) para una lista de objetos *pickable* * En *multiprocessing* tenemos la función `cpu_count` para determinar el número de cores que el sistema operativo puede usar. Este número es la cantidad física o simulada (hyperthreading) de cores.<jupyter_code>import multiprocessing
multiprocessing.cpu_count()<jupyter_output><empty_output><jupyter_text>Pool of workers, ver [Using a pool of workers](https://docs.python.org/3/library/multiprocessing.htmlusing-a-pool-of-workers) La clase `Pool` crea un conjunto (*pool*) de procesos tipo *worker* que procesarán las tareas a realizar vía funciones tipo `map` o `apply`. Se hace `map` del *input* hacia los procesadores y se colecta el *output* de éstos. Mientras el `map` se realiza, el proceso que lanzó el map se bloquea hasta que finalicen las tareas (aunque hay [map_async](https://docs.python.org/3.1/library/multiprocessing.htmlmultiprocessing.pool.multiprocessing.Pool.map_async)). El *output* es una lista.**Obs:** el procesamiento de las tareas podríamos hacerlo con la clase `Process` de arriba pero tendríamos que utilizar un ciclo y colectar los resultados.**Comentario:** para un gran número de tareas a realizar utilicen `Pool`, para pocas tareas a realizar (pocas=menos de $10$) utilicen `Process`.<jupyter_code>from multiprocessing import Pool #importamos clase Pool<jupyter_output><empty_output><jupyter_text>1) Hello world!<jupyter_code>def f(dummy):
    return 'hello world!'
    
if __name__ == '__main__':
    pool = Pool(multiprocessing.cpu_count())
    results = pool.map(f,range(multiprocessing.cpu_count()))
    print(results)
    pool.close()    
    pool.join()
def f(dummy):
    return 'hello world!'
    
if __name__ == '__main__':
    num_processes=2
    pool = Pool(num_processes)
    results = pool.map(f,range(num_processes))
    print(results)
    pool.close()    
    pool.join()<jupyter_output>['hello world!', 'hello world!']<jupyter_text>**Con apply:**<jupyter_code>def f():
    return 'hello world!'
    
if __name__ == '__main__':
    num_processes=2
    pool = Pool(num_processes)
    results = [pool.apply(f) for x in range(num_processes)]
    print(results)
    pool.close()    
    pool.join()<jupyter_output>['hello world!', 'hello world!']<jupyter_text>**Podemos usar un [context manager](https://book.pythontips.com/en/latest/context_managers.htmlcontext-managers) para evitar tener líneas `pool.close()`, `pool.join()`**<jupyter_code>def f(dummy):
    return 'hello world!'
    
if __name__ == '__main__':
    num_processes=2
    with Pool(processes=num_processes) as pool:
        results = pool.map(f,range(num_processes))
        print(results)<jupyter_output>['hello world!', 'hello world!']<jupyter_text>2) Pasar múltiples argumentos vía [starmap](https://docs.python.org/3/library/multiprocessing.htmlmultiprocessing.pool.Pool.starmap)<jupyter_code>def f(s):
    return s
    
if __name__ == '__main__':
    num_processes=2
    with Pool(processes=num_processes) as pool:
        results = pool.starmap(f,[('hola',),('mundo',)])
        print(results)<jupyter_output>['hola', 'mundo']<jupyter_text>3) Regla compuesta del rectángulo<jupyter_code>import math
import time
from scipy.integrate import quad
f=lambda x: math.exp(-x**2)
a=0
b=1
n=10**6
h_hat=(b-a)/n<jupyter_output><empty_output><jupyter_text>**Forma secuencial**<jupyter_code>def Rcf(f, a, b, n): #Rcf: rectángulo compuesto para f
    """
    Compute numerical approximation using rectangle or mid-point method in 
    an interval.
    Nodes are generated via formula: x_i = a+(i+1/2)h_hat for i=0,1,...,n-1 and h_hat=(b-a)/n
    Args:
        f (lambda expression): lambda expression of integrand
        a (int): left point of interval
        b (int): right point of interval
        n (int): number of subintervals
    Returns:
        Rcf (float) 
    """
    h_hat=(b-a)/n
    suma_res=0
    for i in range(0,n):
        x=a+(i+1/2.0)*h_hat
        suma_res+=f(x)
    return h_hat*suma_res
start_time = time.time()
aprox=Rcf(f,a,b,n)
end_time = time.time()
secs = end_time-start_time
print("Rcf tomó",secs,"segundos" )
obj, err = quad(f, a, b)
def err_relativo(aprox, obj):
    return math.fabs(aprox-obj)/math.fabs(obj) #obsérvese el uso de la librería math
err_relativo(aprox,obj)<jupyter_output><empty_output><jupyter_text>La flag `-o` nos permite guardar el *output* de `timeit`.<jupyter_code>Rcf_secuencial_timeit = %timeit -n 5 -r 10 -o Rcf(f,a,b,n)<jupyter_output>379 ms ± 5.9 ms per loop (mean ± std. dev. of 10 runs, 5 loops each)<jupyter_text>**Forma en paralelo**<jupyter_code>p=multiprocessing.cpu_count() #número de procesos
ns_p=int(n/p) #número de subintervalos por proceso
              #se asume que n es divisible por p
              #si no se cumple esto, se puede definir 
              #ns_p=int(n/p) habiendo definido n primero
              #y redefinir n como: 
              #n=p*ns_p
print("número de subintervalos:",n)
print("número de subintervalos por proceso:",ns_p)
def Rcf_parallel(mi_id):
    begin=mi_id*ns_p
    end=begin+ns_p
    h_hat=(b-a)/n
    suma_res=0
    for i in range(begin,end):
        x=a+(i+1/2.0)*h_hat
        suma_res+=f(x)
    return suma_res
if __name__ == '__main__':
    start_time=time.time()
    with Pool(processes=p) as pool:
        results = pool.map(Rcf_parallel,range(p))
        aprox=h_hat*sum(results)
    end_time=time.time()
secs = end_time-start_time
print("Rcf_parallel tomó",secs,"segundos" )
err_relativo(aprox,obj)
%%timeit -n 5 -r 10 -o
if __name__ == '__main__':
    with Pool(processes=p) as pool:
        results = pool.map(Rcf_parallel,range(p))
        aprox=h_hat*sum(results)
Rcf_parallel_timeit=_
Rcf_parallel_timeit.average<jupyter_output><empty_output><jupyter_text>4) Ejemplo para pasar múltiples parámetros a una función vía un [generator](https://wiki.python.org/moin/Generators). Las funciones definidas con `lambda` no son `pickeable`. Utilizamos la forma `def f(x):` siguiente para definir a la función.<jupyter_code>def f(x):
    return math.exp(-x**2)
def Rcf_parallel2(t):
    f,a,b,mi_id,n,ns_p = t
    begin=mi_id*ns_p
    end=begin+ns_p
    h_hat=(b-a)/n
    suma_res=0
    for i in range(begin,end):
        x=a+(i+1/2.0)*h_hat
        suma_res+=f(x)
    return suma_res
if __name__ == '__main__':
    start_time=time.time()
    it=((f,a,b,k,n,ns_p) for k in range(p))
    h_hat=(b-a)/n
    with Pool(processes=p) as pool:
        results = pool.map(Rcf_parallel2,it)
        aprox=h_hat*sum(results)
    end_time=time.time()
secs = end_time-start_time
print("Rcf_parallel2 tomó",secs,"segundos" )
err_relativo(aprox,obj)
%%timeit -n 5 -r 10 -o
if __name__ == '__main__':
    it=((f,a,b,k,n,ns_p) for k in range(p))
    with Pool(processes=p) as pool:
        results = pool.map(Rcf_parallel2,it)
        aprox=h_hat*sum(results)

Rcf_parallel2_timeit=_
Rcf_parallel2_timeit.average
def Rcf_parallel3(t):
    f,a,b,mi_id,n,ns_p = t
    begin=mi_id*ns_p
    end=begin+ns_p
    h_hat=(b-a)/n
    sum_res=0
    f_nodes=(f(a+(i+1/2.0)*h_hat) for i in range(begin,end))
    suma_res=sum(f_nodes)
    return suma_res
if __name__ == '__main__':
    start_time=time.time()
    it=((f,a,b,k,n,ns_p) for k in range(p))
    with Pool(processes=p) as pool:
        results = pool.map(Rcf_parallel3,it)
        aprox=h_hat*sum(results)
    end_time=time.time()
secs = end_time-start_time
print("Rcf_parallel3 tomó",secs,"segundos" )
err_relativo(aprox,obj)
%%timeit -n 5 -r 10 -o
if __name__ == '__main__':
    it=((f,a,b,k,n,ns_p) for k in range(p))
    with Pool(processes=p) as pool:
        results = pool.map(Rcf_parallel3,it)
        aprox=h_hat*sum(results)

Rcf_parallel3_timeit=_
Rcf_parallel3_timeit.average<jupyter_output><empty_output><jupyter_text>En la siguiente propuesta, el proceso principal realiza todas las sumas de la evaluación de los subprocesos:<jupyter_code>def Rcf_parallel4(t):
    f,i,a,b,h_hat = t
    f_nodes=f((a+(i+1/2)*h_hat))
    return f_nodes
if __name__ == '__main__':
    start_time=time.time()
    h_hat=(b-a)/n
    it=((f,i,a,b,h_hat) for i in range(n))
    with Pool(processes=p) as pool:
        results = pool.map(Rcf_parallel4,it)
    suma_res=sum(results)
    aprox=h_hat*suma_res
    end_time=time.time()
secs = end_time-start_time
print("Rcf_parallel4 tomó",secs,"segundos" )
err_relativo(aprox,obj)
%%timeit -n 5 -r 10 -o
if __name__ == '__main__':
    it=((f,i,a,b,h_hat) for i in range(n))
    with Pool(processes=p) as pool:
        results = pool.map(Rcf_parallel4,it)
    suma_res=sum(results)
    aprox=h_hat*suma_res

Rcf_parallel4_timeit=_
Rcf_parallel4_timeit.average<jupyter_output><empty_output><jupyter_text>Gráfica de tiempo de ejecución vs número de procesos<jupyter_code>import matplotlib.pyplot as plt
(Rcf_secuencial_timeit.average,Rcf_parallel_timeit.average,
 Rcf_parallel2_timeit.average,Rcf_parallel3_timeit.average,
 Rcf_parallel4_timeit.average)<jupyter_output><empty_output><jupyter_text>El mejor tiempo lo tiene `Rcf_parallel2` con:<jupyter_code>p<jupyter_output><empty_output><jupyter_text>procesos. Usaremos ésta implementación para realizar la gráfica variando el número de procesos de $1$ hasta `multiprocessing.cpu_count()`<jupyter_code>multiprocessing.cpu_count()
err_np=[]
n_cpus=[]
n=10**6 #cambiar n si se desea
h_hat=(b-a)/n
def mifun(p,ns_p):
    it=((f,a,b,k,n,ns_p) for k in range(p))
    with Pool(processes=p) as pool:
        results = pool.map(Rcf_parallel2,it)
        aprox=h_hat*sum(results)
    return err_relativo(aprox,obj)
for p in range(1,multiprocessing.cpu_count()+1):
    if n%p==0:
        ns_p=int(n/p)
        err_np.append(mifun(p,ns_p))
        n_cpus.append(p)
err_np
n_cpus
l=[]
n_cpus=[]
for p in range(1,multiprocessing.cpu_count()+1):
    if n%p==0:
        ns_p=int(n/p)
        resultado_timeit=%timeit -n 5 -r 10 -o mifun(p,ns_p)
        l.append(resultado_timeit.average)
        n_cpus.append(p)
n_cpus
l
plt.plot(n_cpus,l,'o-')
plt.title('Gráfica num cpus vs tiempo')
plt.xlabel('num cpus')
plt.ylabel('tiempo')
plt.grid()
plt.show()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12092875318066158, 'fraction_numerical': 0.012150127226463104, 'mean_word_length': 4.648939992813511, 'pattern_counts': {'":': 0, '<': 63, '<?xml version=': 0, '>': 63, 'https://': 18, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '17', 'dir': 'jupyter-structured-clean-dedup', 'id': '428296', 'license': 'Apache-2.0', 'n_tokens_mistral': 5990, 'n_tokens_neox': 5877, 'n_words': 1434, 'path': 'temas/II.computo_paralelo/2.2.Python_multiprocessing.ipynb', 'repo_name': 'AliciaMstt/analisis-numerico-computo-cientifico', 'size': '15716'}