Skip to main content
simplificado el código para que se entienda algo mejor
Origen Enlace
ChemaCortes
  • 8.9k
  • 22
  • 36
import randomtime
from collections import deque
from itertools import islice
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, as_completed

def fact(n):
    res = 1
    for i in range(2, n+1):
        res *= i
    return res

if __name__ == "__main__":

    with ProcessPoolExecutor() as e:
        # metemos 1000 procesos
        futuresfs1 = { e.submit(fact, i):i 
                    for i in (random.randrange(10000) for x in range(1000)) }
        print("Lanzados 1000 futuros")

        # obtenemos los primeros 10 resultados
        res10 = { futures[f]fs1[f]:f.result() for f in islice(as_completed(futuresfs1), 10)}
        
        # metemos otros 1000100000 más
        futures2fs2 = { e.submit(fact, i):i 
          for i in range(1000, 100000) }
     for i in (random.randrangeprint(10000) for"Lanzados x100000 infuturos range(1000)más") }

        # obtenemos los siguientes 10 resultados de la segunda tanda
        res20 = { futures2[f]fs2[f]:f.result() for f in islice(as_completed(futures2fs2), 101000, 1010)}

        # obtenemos los últimos 10 resultados de la tanda anterior
        res30 = { futures[f]fs1[f]:f.result() for f in deque(as_completed(futuresfs1), 10)}
        
        # salimosEspera delpara contextocomprobar cómo sube el consumo de ejecución,CPU
 con lo      print("Espera de 3 segundos")
    # que terminan  time.sleep(3)
        
        # Cancelamos todos los procesos que no hayan acabado de la segunda tanda
        for f in fs2:
            f.cancel()
        print("Cierre del ejecutor de procesos")

    print()
    for (n,res) in res10.items():
        print(f"fact({n:4d})) = {str(res)[:80]}...")
    
    print()
    for (n,res) in res20.items():
        print(f"fact({n:4d})) = {str(res)[:80]}...")
    
    print()
    for (n,res) in res30.items():
        print(f"fact({n:4d})) = {str(res)[:80]}...")

Edición: he simplificado el código para que se entienda algo mejor.

import random
from collections import deque
from itertools import islice
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, as_completed

def fact(n):
    res = 1
    for i in range(2, n+1):
        res *= i
    return res

if __name__ == "__main__":

    with ProcessPoolExecutor() as e:
        # metemos 1000 procesos
        futures = { e.submit(fact, i):i 
                    for i in (random.randrange(10000) for x in range(1000)) }
        
        # obtenemos los primeros 10 resultados
        res10 = { futures[f]:f.result() for f in islice(as_completed(futures), 10)}
        
        # metemos otros 1000 más
        futures2 = { e.submit(fact, i):i 
                    for i in (random.randrange(10000) for x in range(1000)) }

        # obtenemos los siguientes 10 resultados
        res20 = { futures2[f]:f.result() for f in islice(as_completed(futures2), 10)}

        # obtenemos los últimos 10 resultados de la tanda anterior
        res30 = { futures[f]:f.result() for f in deque(as_completed(futures), 10)}
        
    # salimos del contexto de ejecución, con lo
    # que terminan todos los procesos
    for (n,res) in res10.items():
        print(f"fact({n})) = {str(res)[:80]}...")
    for (n,res) in res20.items():
        print(f"fact({n})) = {str(res)[:80]}...")
    for (n,res) in res30.items():
        print(f"fact({n})) = {str(res)[:80]}...")
import time
from collections import deque
from itertools import islice
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, as_completed

def fact(n):
    res = 1
    for i in range(2, n+1):
        res *= i
    return res

if __name__ == "__main__":

    with ProcessPoolExecutor() as e:
        # metemos 1000 procesos
        fs1 = { e.submit(fact, i):i for i in range(1000) }
        print("Lanzados 1000 futuros")

        # obtenemos los primeros 10 resultados
        res10 = { fs1[f]:f.result() for f in islice(as_completed(fs1), 10)}

        # metemos otros 100000 más
        fs2 = { e.submit(fact, i):i for i in range(1000, 100000) }
        print("Lanzados 100000 futuros más")

        # obtenemos 10 resultados de la segunda tanda
        res20 = { fs2[f]:f.result() for f in islice(as_completed(fs2), 1000, 1010)}

        # obtenemos los últimos 10 resultados de la tanda anterior
        res30 = { fs1[f]:f.result() for f in deque(as_completed(fs1), 10)}
        
        # Espera para comprobar cómo sube el consumo de CPU
        print("Espera de 3 segundos")
        time.sleep(3)
        
        # Cancelamos todos los procesos que no hayan acabado de la segunda tanda
        for f in fs2:
            f.cancel()
        print("Cierre del ejecutor de procesos")

    print()
    for (n,res) in res10.items():
        print(f"fact({n:4d}) = {str(res)[:80]}...")
    
    print()
    for (n,res) in res20.items():
        print(f"fact({n:4d}) = {str(res)[:80]}...")
    
    print()
    for (n,res) in res30.items():
        print(f"fact({n:4d}) = {str(res)[:80]}...")

Edición: he simplificado el código para que se entienda algo mejor.

Corregir la identación del código
Origen Enlace
ChemaCortes
  • 8.9k
  • 22
  • 36

import random from collections import deque from itertools import islice from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, as_completed

def fact(n): res = 1 for i in range(2, n+1): res *= i return res

if name == "main":

import random
from collections import deque
from itertools import islice
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, as_completed

def fact(n):
    res = 1
    for i in range(2, n+1):
        res *= i
    return res

if __name__ == "__main__":

    with ProcessPoolExecutor() as e:
        # metemos 1000 procesos
        futures = { e.submit(fact, i):i 
                    for i in (random.randrange(10000) for x in range(1000)) }
        
        # obtenemos los primeros 10 resultados
        res10 = { futures[f]:f.result() for f in islice(as_completed(futures), 10)}
        
        # metemos otros 1000 más
        futures2 = { e.submit(fact, i):i 
                    for i in (random.randrange(10000) for x in range(1000)) }

        # obtenemos los siguientes 10 resultados
        res20 = { futures2[f]:f.result() for f in islice(as_completed(futures2), 10)}

        # obtenemos los últimos 10 resultados de la tanda anterior
        res30 = { futures[f]:f.result() for f in deque(as_completed(futures), 10)}
        
    # salimos del contexto de ejecución, con lo
    # que terminan todos los procesos
    for (n,res) in res10.items():
        print(f"fact({n})) = {str(res)[:80]}...")
    for (n,res) in res20.items():
        print(f"fact({n})) = {str(res)[:80]}...")
    for (n,res) in res30.items():
        print(f"fact({n})) = {str(res)[:80]}...")

import random from collections import deque from itertools import islice from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, as_completed

def fact(n): res = 1 for i in range(2, n+1): res *= i return res

if name == "main":

with ProcessPoolExecutor() as e:
    # metemos 1000 procesos
    futures = { e.submit(fact, i):i 
                for i in (random.randrange(10000) for x in range(1000)) }
    
    # obtenemos los primeros 10 resultados
    res10 = { futures[f]:f.result() for f in islice(as_completed(futures), 10)}
    
    # metemos otros 1000 más
    futures2 = { e.submit(fact, i):i 
                for i in (random.randrange(10000) for x in range(1000)) }

    # obtenemos los siguientes 10 resultados
    res20 = { futures2[f]:f.result() for f in islice(as_completed(futures2), 10)}

    # obtenemos los últimos 10 resultados de la tanda anterior
    res30 = { futures[f]:f.result() for f in deque(as_completed(futures), 10)}
    
# salimos del contexto de ejecución, con lo
# que terminan todos los procesos
for (n,res) in res10.items():
    print(f"fact({n})) = {str(res)[:80]}...")
for (n,res) in res20.items():
    print(f"fact({n})) = {str(res)[:80]}...")
for (n,res) in res30.items():
    print(f"fact({n})) = {str(res)[:80]}...")
import random
from collections import deque
from itertools import islice
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, as_completed

def fact(n):
    res = 1
    for i in range(2, n+1):
        res *= i
    return res

if __name__ == "__main__":

    with ProcessPoolExecutor() as e:
        # metemos 1000 procesos
        futures = { e.submit(fact, i):i 
                    for i in (random.randrange(10000) for x in range(1000)) }
        
        # obtenemos los primeros 10 resultados
        res10 = { futures[f]:f.result() for f in islice(as_completed(futures), 10)}
        
        # metemos otros 1000 más
        futures2 = { e.submit(fact, i):i 
                    for i in (random.randrange(10000) for x in range(1000)) }

        # obtenemos los siguientes 10 resultados
        res20 = { futures2[f]:f.result() for f in islice(as_completed(futures2), 10)}

        # obtenemos los últimos 10 resultados de la tanda anterior
        res30 = { futures[f]:f.result() for f in deque(as_completed(futures), 10)}
        
    # salimos del contexto de ejecución, con lo
    # que terminan todos los procesos
    for (n,res) in res10.items():
        print(f"fact({n})) = {str(res)[:80]}...")
    for (n,res) in res20.items():
        print(f"fact({n})) = {str(res)[:80]}...")
    for (n,res) in res30.items():
        print(f"fact({n})) = {str(res)[:80]}...")
Origen Enlace
ChemaCortes
  • 8.9k
  • 22
  • 36

A pesar del comentario que he puesto antes, voy a intentar dar un ejemplo usando Futures:

Los "futures" son una abstracción para la ejecución concurrente de código que igual sirven para la ejecución de hilos (Threads) que para la ejecución de procesos (Process). Para ello utiliza un gestor de ejecución que es quien se encarga de la ejecución y de entregar los resultados (las "promesas") a quien las solicite.

Por ejemplo, queremos calcular el factorial de una secuencia de números. Con

with executor as e:
    futures = {e.submit(fact, i):i for i in range(0,4000)}

lanzaríamos la ejecución en paralelo del cálculo del factorial de 4000 números. El executor es quien gestiona hilos (ThreadPoolExecutor) o procesos (ProcessPoolExecutor), incluso podemos dejar que decida el número de "workers" que debe emplear según sea el número de cores de nuestra CPU.

El ejemplo completo:

import time
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, wait

def fact(n):
    res = 1
    for i in range(2, n+1):
        res *= i
    return res

if __name__ == "__main__":

    t0 = time.time()
    
    with ThreadPoolExecutor(max_workers=50) as e:
        fs = {e.submit(fact, i):i for i in range(4000)}
        wait(fs)
    
    t1 = time.time()
    
    with ProcessPoolExecutor() as e:
        fs = {e.submit(fact, i):i for i in range(4000)}
        wait(fs)
    
    t2 = time.time()
    
    print(f"Ejecución con hilos: {t1-t0:.2f}s")
    print(f"Ejecución con procesos: {t2-t1:.2f}s")

Por lo general, los tiempos en windows con procesos son bastante malos, pudiendo llegar a ser peores que con hilos en algunos casos.

Un modo de trabajar "asíncronamente" sería ir metiendo al ejecutor más tareas sobre la marcha, algo así:

import random from collections import deque from itertools import islice from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, as_completed

def fact(n): res = 1 for i in range(2, n+1): res *= i return res

if name == "main":

with ProcessPoolExecutor() as e:
    # metemos 1000 procesos
    futures = { e.submit(fact, i):i 
                for i in (random.randrange(10000) for x in range(1000)) }
    
    # obtenemos los primeros 10 resultados
    res10 = { futures[f]:f.result() for f in islice(as_completed(futures), 10)}
    
    # metemos otros 1000 más
    futures2 = { e.submit(fact, i):i 
                for i in (random.randrange(10000) for x in range(1000)) }

    # obtenemos los siguientes 10 resultados
    res20 = { futures2[f]:f.result() for f in islice(as_completed(futures2), 10)}

    # obtenemos los últimos 10 resultados de la tanda anterior
    res30 = { futures[f]:f.result() for f in deque(as_completed(futures), 10)}
    
# salimos del contexto de ejecución, con lo
# que terminan todos los procesos
for (n,res) in res10.items():
    print(f"fact({n})) = {str(res)[:80]}...")
for (n,res) in res20.items():
    print(f"fact({n})) = {str(res)[:80]}...")
for (n,res) in res30.items():
    print(f"fact({n})) = {str(res)[:80]}...")

Hay que destacar que res20 almacena resultados únicamente de la segunda tanda de futuros futures2. En ese instante podríamos haber cancelado la ejecución de los futuros pendientes de la primera tanda si no se iban a necesitar para nada más.

PD: si no te funciona, asegúrate que usas python 3.6