Programación concurrente

PROGRAMACIÓN ESTRUCTURADA

La programación estructurada no es más que un paradigma de programación.

En términos simples la programación estructurada tiene como objetivo mejorar el desarrollo de software recurriendo a únicamente subrutinas y tres estructuras básicas: Secuencial, Selectiva e Iterativa.

En terminos de programación una estructura selectiva hace referencia a condicionales, por ejemplo, un if, else, else if o swith. Para una estructura iterativa estaremos hablando de ciclos, por ejemplo for, foreach, while o do while.

Lo que nos interesa para este post es la estructura secuencial. En programación, una estructura secuencial no es más que la ejecución, en secuencia, de múltiples tareas. Estas tareas serán ejecutadas una tras otra, todas en un orden previamente definido. ☝ Una tarea no podrá ejecutarse antes de otra si en el programa no fue indicado de esa manera.

Ejemplo.

print('Hola, soy un programa muy simple')

nombre = input('¿Cual es tu nombre? ')

print('Mucho gusto ' + nombre)

print('Adios, fin del programa')

En este ejemplo podemos decir que tenemos cuatro tareas, una por cada línea de código. El orden de ejecución es descendente.

El programa no puede imprimir en consola el nombre del usuario sin antes haberlo pedido. De igual forma, el programa no puede despedirse sin antes haber dado la bienvenida. Podemos concluir que este programa es secuencial, ya que cada tarea se ejecuta una tras otra, en un orden.

CONCURRENCIA Y PARALELISMO

Estos conceptos en muchas ocasiones se confunden y se puede llegar a pensar que se tratan de lo mismo, cuando en realidad no es así, veamos.

La concurrencia es, en esencia, el poder realizar varias tareas al mismo tiempo, pero no especificamente en paralelo. — wait, What?

Una de las formas más sencillas de comprender la concurrencia es imaginar a una persona la cual trabaja en múltiples tareas al mismo tiempo, y que rápidamente cambia de una tarea a otra.

Concluimos que la persona trabaja de forma concurrente. Las tareas que realiza no necesariamente deben seguir un orden.

Al contrario que una estructura secuencial, con la concurrencia el orden en que se ejeuctan las tareas importa muy poco.

Ejemplo de concurrencia.

import time
import threading

def codificar():
    time.sleep(2)

    print(f'Codificando')

def responder_correos():
    time.sleep(2)

    print(f'Respondiendo correos')

def realizar_calculos():
    time.sleep(2)

    print(f'Realizar los calculos')

threading.Thread(target=codificar).start()
threading.Thread(target=responder_correos).start()
threading.Thread(target=realizar_calculos).start()

En este caso nuestro programa realiza tres tareas al mismo tiempo. A cada tarea le tomó un máximo de dos segundos ser completada, como se ejecutan de forma concurrente (al mismo tiempo) al programa le toma dos segundo finalizar su ejecución.

Por otro lado, si ejecutamos el mismo código, pero ahora de forma secuencial, al programa le tomaría seis segundos finalizar. El tiempo es sin duda considerable.

#secuencial

codificar()
responder_correos()
realizar_calculos()

El paralelismo es el poder de ejecutar dos o más acciones de forma simultánea, en lugar de concurrentemente.

Si recordamos, en nuestro ejemplo anterior, el desarrollador realiza tres tareas al mismo tiempo, realizaba calculos programaba y contestaba correos, sin embargo, ninguna de estas tareas se realizaba de forma simultánea.

En nuestro ejemplo, si queremos que las tareas se realicen de forma paralela tendríamos que tener a tres personas trabajando, vaya, una persona por cada tarea, una persona encargada de los cálculos, otra de la codificación y otra respondiendo correos, tres personas trabajando simultáneamente.

Técnicamente hablando, para llevar a cabo un verdadero paralelismo es necesario que nuestra computadora posea múltiples procesadores, con esto, cada tarea será ejecutada en un procesador diferente, de forma simultánea, de forma paralela.

Para crear paralelismo con Python será necesario utilizar el modulo processing.

import os
import time
import logging
import multiprocessing

logging.basicConfig(level=logging.DEBUG, format='%(message)s')

def new_process(time_to_sleep=0):
    logging.info('Comenzamos el proceso hijo!')

    time.sleep(time_to_sleep)

    logging.info('Terminamos el procesos hijo!')

def main():
    process = multiprocessing.Process(target=new_process,
                                        name='proceso-hijo',
                                        args=(1,),
                                        daemon=False)
    process.start()
    process.join()

    logging.info(f'Finalizamos el proceso principal')

if __name__ == '__main__':
    main()

La principal diferencia entre concurrencia y paralelismo recae en la forma en que se realizan las tareas. Cuando se ejecutan tareas de forma concurrente a estas se les asigna un x periodo de tiempo antes de cambiar de tarea, será en ese periodo en el cual se inicie, continúe, o se complete la tarea, por otro lado, si ejecutamos tareas en paralelo, las tareas se realizarán de forma simultánea, comenzarán y finalizarán sin interrupciones.

Si deseamos implementar concurrencia en nuestros programas una muy buena idea será utilzar Threads, por otro lado, si deseamos implementar paralelismos optaremos por procesos.

En conclusión y en términos simples, la programación concurrente no es más que la forma en la cual podemos resolver ciertas problemáticas de forma concurrente, es decir, ejecutando múltiples tareas a la misma vez y no de forma secuencial.

En un programa concurrente las tareas puede continuar sin la necesidad que otras comiencen o finalicen.

Si bien es cierto que la programación concurrente acarrea ciertos problemas, principalmente al momento de compartir información entre tareas, también es cierto que si se implementa de forma correcta, podremos, en casos puntuales, mejorar significativamente el performance de nuestras aplicaciones.

Threads y Procesos.

PROCESOS

Un proceso no es más que la instancia de un programa. Verás, cuando tú abres cualquier programa en tu dispositivo, por ejemplo, tu navegador web, quizás google chrome, internamente estás creando un nuevo proceso.

Un proceso no será más que la ejecución del programa mismo. Y ojo, no hay que confundir un proceso con un programa ya que son entidades completamente diferentes.

Podemos ver a un programa como un conjunto de instrucciones y de datos, por otro lado, podemos ver a un proceso como la ejecución de esas instrucciones junto con esos datos.

Una muy buena analogía es ver a un programa como la receta para preparar un platillo, ya que en una receta se establecen todos los ingredientes a utilizar junto con los pasos a seguir; en cambio, un proceso podemos verlo como la ejecución de esa receta, algo que se está realizando.

Puede quedarnos más en claro si consideramos que un mismo programa puede ser ejecutado múltiples veces por nuestro sistema operativo. Siguiendo con la analogía, una misma recete puede ser implementada por n chefs.

Dentro de un proceso encontraremos todo lo necesario para que el programa se ejecute de forma correcta, me refiero a código fuente, ficheros, variables, tareas, sub-procesos etc...

Una de las tareas del sistema operativo es aislar a cada uno de los procesos entre sí, de tal forma que los procesos sean completamente independientes unos de otros; esto con la finalidad prevenir que compartan información entre ellos, lo cual pueda dar pie a errores. Es por ello que en esencia, lo programas no pueden acceder a la información y datos de otros programa en ejecución.

Cuando un proceso finaliza, ya sea de forma natural o no, quizás por algún error, será el sistema operativo el encargado de liberar el espacio en memoria.

Podemos ver al sistema operativo como un maestro de ceremonias, quien está al pendiente de todos los procesos, desde su creación, ejecución y finalización.

Con los procesos es posible implementar el paralelismo, ejecutando diferentes procesos en diferentes procesadores, claro, tambíen depende de cuantos procesadores posea el equipo de computo.

Para que un proceso pueda ser ejecutado este necesita poseer por lo menos un thread, a este thread lo vamos a conocer como el thread principal, o, el main thread.

Un proceso no es más que la ejecución un programa.

THREADS

Los Threads, también conocidos como sub-procesos o hilos, podemos definirlos como una secuencia de instrucciones las cuales el sistema operativo puede programar para su ejecucción.

A diferencia de un proceso, los threads son entidades mucho más pequeñas, lo cual los hace fácilles de gestionar, tanto es así que un thread es la unidad más pequeña a la cual un procesador puede asignar tiempo.

A diferencias de los procesos, los cuales viven dentro del sistema operativo, los threads viven dentro de los procesos. Un thread se crean, ejecutan y finalizan dentro de un proceso, dicho en otras palabras: Un thread le pertenece a un proceso, y, asu vez, un proceso puede poseer múltiples threads. Si lo vemos en términos de base de datos pudiésemos decir que esta es una relación uno a muchos.

Un Proceso es una composicion de unos o mas threads.

Algo interesante a mencionar es que debido a que los Thread existen dentro de los procesos, estos pueden compartir información entre ellos; Algo a lo cual sin duda le podemos sacar mucho provecho, pero, que también puede llegar hacer complicado de manejar.

Uno de los problemas más comunes al momento de trabajar con Threads tiene por nombre Race condition, y es, en esencia, un problema que surge cuando más de un thread intenta acceder y modificar un espacio en memoria compartido, ocasionando que el programa se comporte de forma inadecuada.

Visto de otra forma, un thread no es más que un proceso, pero en pequeño, al igual que un proceso un thread se ejecuta dentro de un contexto y posee instrucciónes a realizar

Un thread es una secuencia de instrucciones las cuales preparan la ejecucion de una accion (proceso). Estos se crean, se ejecutan y mueren dentro de los procesos.

Con los threads y los procesos seremos capaces de implementar la programación concurrente, y, dependiendo de la cantidad de procesadores la programación en paralelo.

1-threads.py

import requests
def get_name():
    response = requests.get('https://randomuser.me/api/')
    if response.status_code == 200:
        
        results = response.json().get('results')
        name = results[0].get('name').get('first')

        print(name)

if __name__ == "__main__":
    #Secuencial
    for _ in range(0, 20):
        get_name()

Si se ejecuta el codigo tal cual como está, cargara lentamente uno por uno los nombres, esperando la repuesta del servidor por cada iteracion hasta satisfacer los 20 nombres por los cuales itera nuestro bucle for.

En cambio si se usan threads.

import requests
import threading

def get_name():
    response = requests.get('https://randomuser.me/api/')
    if response.status_code == 200:
        
        results = response.json().get('results')
        name = results[0].get('name').get('first')

        print(name)

if __name__ == "__main__":
    #Concurrente
    for _ in range(0 ,20):
        thread = threading.Thread(target=get_name)
        thread.start()

Ahora los nombres se muestran en pantalla más rapido, debido a que el programa realiza peticione de forma concurrente. O sea, no es necesario que el servidor responda para realizar la siguiente petición.

De forma concurrente mejora el performance del script.

En este modulo haremos uso de la libreria threading.

import threading

Apoyandonos de su clase Thread.

thread = threading.**Thread**(target=entity)

target recibe como valor aquella entidad la cual queremos que se ejecute de forma concurrente.

No basta con generar el thread, debemos programar su arranque.

thread.start()

Enviar argumento a una tarea

import threading

def executor_a():
    for x in range(0,10):
        print(f'Hi!, I\'m the thread n°1, iteration {x}')

def executor_b():
    for x in range(0,10):
        print(f'Hi!, I\'m the thread n°2, iteration {x}')

def executor_c():
    for x in range(0,10):
        print(f'Hi!, I\'m the thread n°3, iteration {x}')

thread_a = threading.Thread(target=executor_a)
thread_b = threading.Thread(target=executor_b)
thread_c = threading.Thread(target=executor_c)

thread_a.start()
thread_b.start()
thread_c.start()

En este pedazo de codigo encontramos que cada executor_a,b,c es ejecutado en un thread.

La salida de este es algo así.

Hi!, I'm the thread n°2, iteration 6
Hi!, I'm the thread n°1, iteration 8
Hi!, I'm the thread n°2, iteration 7
Hi!, I'm the thread n°1, iteration 9
Hi!, I'm the thread n°2, iteration 8
Hi!, I'm the thread n°2, iteration 9

Si te fijas, no hay orden en la ejecucción, esto se debe a los principios de concurrencia, este fenomeno se conoce como alternancia.

Un caso muy común es que las tareas que queremos ejecutar de forma concurrente reciban valores de entrada.(o sea, que posean parametros)

Para ello es posible enviar argumentos antes de ejecutar las funciones.

Para esto hay dos formas.

def executor_a(id=0):
    for x in range(0,10):
        print(f'Hi!, I\'m the thread {id}, iteration {x}')

thread_a = threading.Thread(target=executor_a, **args=[1]**)

Esta primera forma da el argumento dentro de una lista como parametro de la funcion executor_a.

La salida de esto es:

Hi!, I'm the thread 1, iteration 1
Hi!, I'm the thread 1, iteration 2
Hi!, I'm the thread 1, iteration 3
...
Hi!, I'm the thread 1, iteration 20

U otra es usar una tupla:

def executor_a(id=0):
    for x in range(0,10):
        print(f'Hi!, I\'m the thread {id}, iteration {x}')

thread_a = threading.Thread(target=executor_a, **args=(1,)**)

Siempre colocar la "," despues de cada elemento de al tupla.

Esto retorna lo siguiente:

Hi!, I'm the thread 1, iteration 1
Hi!, I'm the thread 1, iteration 2
Hi!, I'm the thread 1, iteration 3
...
Hi!, I'm the thread 1, iteration 20

La forma número 2 hace uso del el keyword kwargs y uso de un diccionario***:***

thread_a = threading.Thread(target=executor_c, **kwargs={'id':1}**)

Aquí se usa como key del diccionario la variable asignada como en el parametro de la función. (id)

Modulo Logging

En algun punto clave de algun proyecto solemos (siempre) usar la funcion print, esta no suele ser muy efectiva cuando utilizamos Threads.

Esto principalmente porque existen otras herramientas que nos permitiran testear nuestras applicaciones con mensajes dentro del terminal.

Para comenzar debemos importar la libreria logging:

import logging

El modulo logging nos ofrece 5 tipos de mensajes.

Debug (10)
Info (20)
Warning (30)
Error (40)
Critical (50)

Estos tipos poseen un nivel, a partir de estos podemos regular que mensajes mostrar y cuales no.

import logging

logging.debug('This is a Debug message')
logging.info('This is a Info message')    
logging.warning('This is a Warning message')
logging.error('This is a Error message')
logging.critical('This is a Critical message')

Al ejecutar este codigo devuelve lo siguiente:

WARNING:root:This is a Warning message
ERROR:root:This is a Error message
CRITICAL:root:This is a Critical message

Que?, le faltan 2 mensajes más.

Esto sucede debido a que la libreria viene configurada para imprimir en consola mensajes desde el nivel 30 hacia arriba.

¿Solucion?

import logging

logging.basicConfig(
        level=10
)

Aqui se imprimiran mensajes desde el nivel 10.

Otra forma similar seria envéz de colocar el número 10, colocar una constante.

logging.basicConfig(
				level=**logging.DEBUG** #10
)

Si te fijas, los mensajes impresos por el módulo logging tienen un formato por default, este se puede modificar para que en pequeñas ocasiones contenga un formato distinto.

logging.basicConfig(
        level=10,
        format='%(filename)s - %(asctime)s'
)

Esto arroja:

3.-logging.py - 2020-05-28 23:15:52,307
3.-logging.py - 2020-05-28 23:15:52,308
3.-logging.py - 2020-05-28 23:15:52,308
3.-logging.py - 2020-05-28 23:15:52,309
3.-logging.py - 2020-05-28 23:15:52,309

filename arroja el nombre del archivo, asctime la hora exacta

Como ves el formato es el siguiente: '%( )s'

Hay que destacar que el "-" que está entre los dos formatos, es opcional, este puede ser un guion, una letra, una coma, etc..

Otra cosa interesante es que cuando trabajamos con fechas podemos modificar su formato de la siguiente forma.

logging.basicConfig(
        level=10,
        format='%(filename)s - %(asctime)s',
        datefmt='%H:%M:%S'
)

Salida:

3.-logging.py - 23:23:28
3.-logging.py - 23:23:28
3.-logging.py - 23:23:28
3.-logging.py - 23:23:28
3.-logging.py - 23:23:28

Ademas podemos mostrar el la funcion en ejecucion:

logging.basicConfig(
        level=10,
        format='%(message)s - %(funcName)s'
)

Salida:

This is a Debug message - mi_funcion
This is a Info message - mi_funcion

Tambien podemos mostrar la liena que se ejecuta, el nivel del mensaje y la ruta.

logging.basicConfig(
        level=10,
        format='%(message)s - %(lineno)s - %(levelname)s - %(pathname)s'
)

Output:

This is a Debug message - 14 - DEBUG - 3.-logging.py
This is a Info message - 15 - INFO 3.-logging.py
This is a Warning message - 16 - WARNING 3.-logging.py
This is a Error message - 17 - ERROR 3.-logging.py
This is a Critical message - 18 - CRITICAL 3.-logging.py

Pero el uso realmente importante es que nos muestre un mensaje del thread en uso:

logging.basicConfig(
        level=10,
        format='%(message)s - %(thread)s - %(threadName)s'
)

En la salida nos muestra el thread y su nombre.

3.-logging.py - 139650562291520 - MainThread
3.-logging.py - 139650562291520 - MainThread

De igual forma el proceso y su nombre.

logging.basicConfig(
        level=10,
        format='%(process)s - %(processName)s'
)

Esto arroja:

812 - MainProcess
812 - MainProcess

Otra cosa mucho más interesante es que los mensajes, si no queremos que se muestren en pantalla, podemos guardarlos en un archivo.

logging.basicConfig(
        level=10,
        format='%(filename)s - %(asctime)s',
        datefmt='%H:%M:%S',
        filename='message.txt'
)

Aqui estamos enviando todos los mensajes a un archivo de texto, estos no se muestran en la terminal.

Al abrir el archivo messages.txt este contiene:

3.-logging.py - 23:25:56
3.-logging.py - 23:25:56
3.-logging.py - 23:25:56
3.-logging.py - 23:25:56
3.-logging.py - 23:25:56

Thread Principal

Algo a tener en cuenta es que todos los procesos se ejecutan en el thread principal, o más bien conocido como main thread

logging.basicConfig(
        level=logging.DEBUG,
        format='%(thread)s %(threadName)s : %(message)s'
)

140463854307136 MainThread : Estamos en el Main Thread

Dormir Threads

Al dormir threads lo que hacemos es odernarle al ordenador que nuestro thread no estara activo durante una cierta cantidada de tiempo.

Esto da el paso a que otros threads tomen su lugar en el procesador.

Para dormir un Thread usando python, necesitamos una libreria llamada time.

import time
import logging
import threading

def task():
		logging.warning('Antes de que se duerma')
		time.sleep(2)
		logging.warning('Se durmio y desperto!')

if __main__ == '__name__':
		thread = threading.Thread(target=task)
		thread.start()

Como puedes leer, del modulo time nosotros usamos la clase sleep, esta en sus parametros requiere segundos.

#time.sleep(segundos)

Lo que arroja el codigo anterior es lo siguiente.

WARNING:root:Antes de que se duerma
WARNING:root:Se durmio y desperto

Al terminar la primera linea se toma 2 segundos y luego escribe la otra, eso debido porque es thread se fue a dormir.

Callbacks

Son funciones que enviamos como argumento a otras funciones.

import logging
import requests
import threading

logging.basicConfig(level=10, format='%(message)s')

def get_pokemon_name(response_json):
        name = response_json.get('forms')[0].get('name')
        logging.info(f'Nombre:{name}')

def error():
        pass
#Hollywood no nos llames, nostros te llamamos
def generate_request(url, success_callback, error_callback)
        response = requests.get(url)

        if response.status_code == 200:
                success_callback(response.json())

        else:
                error_callback()

if __name__ == '__main__':
        thread1 = threading.Thread(target=generate_request,
																	 kwargs={'url':'https://pokeapi.co/api/v2/pokemon/1',
																					 'success_callback':get_pokemon_name,
																					 'error_callback':error})
 
        thread1.start()

Lo que sucede aquí es que el thread ejecuta la funcion generate_request, esta le pasa como parametros los callbacks definidos en kwargs del target, luego de que el servidor responda la llamada, es derivado a success_callback() o error_callback(), desde ahi vuelve al thread y va a las funciones llamadas con el callback, serian get_pokemon_name o error.

Lo principal por lo que se usan los callbacks es debido a que ya tenemos una funcion como target, y en esa hacemos correr mas funciones en sus parametros.

Programar Callbacks

En esta seccion usaremos el modulo timer proveniente de el module threading.

Lo que hace esta esta clase es arrancar un callback luego del periodo de tiempo especificado.

import threading
import logging

def callback():
		logging.warning('Soy un callback lento')

if __name__ == '__main__':
		thread = threading.Timer(3, callback)
		thread.start()

		logging.warning('Soy el Main Thread')
		logging.warning('Soy el más rapido')

Si pruebas esto devuelve lo siguiente:

WARNING:root:Soy el Main Thread
WARNING:root:Soy el más rapido
(luego de 3 segundos)
WARNING:root:Soy un callback lento

El formato de trabajar con la clase timer es el siguiente:

#threading.Timer(segundos, funcion)

Futuros

Un futuro es la abstracción de un resultado, no podremos hacer uso de ese resultado hasta dentro de un futuro. En otras palabras, representa un valor eventual.

import logging
from concurrent.futures import Future

logging.basicConfig(level=10, format='%(message)s')

def callback_futuro(result):
				logging.info('Me ejecuto cuando el futuro tenga un valor!')

if __name__ == '__main__':
        future = Future()
				future.add_done_callback(callback_futuro)

				logging.info('''Hola, date cuenta que el futuro tiene callback, 
												pero no tiene un valor''')

				future.set_result('Soy un valor')

				logging.info('Ahora el futuro tiene un valor!!')

from concurrent.futures import Future: Esto trae la clase Future

future = Future(): Generamos la instancia de la clase, para poder usarla.

definimos el callback y le damos como parametro result, el cual definiremos posteriormente.

future.set_result('Soy un valor'): Aquí se le da un valor a nuestro futuro con set_result.

Hola, date cuenta que el futuro tiene callback, pero no tiene un valor
Aun no posee valor
Me ejecuto cuando el futuro tenga un valor!
Ahora el futuro tiene un valor!!

Si te fijas, no se muestra en pantalla el result, solo el callback.

En resumidas cuentas actua ejecutando el callback cuando se tiene un valor, esto se puede aplicar en algunos casos por un bloque condicional.

Una vez tengamos un valor se ejecutaran todos los callbacks. (si plural)

COMO, AGREGAR MÁS DE UNO?

import logging
from concurrent.futures import Future

logging.basicConfig(level=10, format='%(message)s')

def callback_futuro(result):
				logging.info('Me ejecuto cuando el futuro tenga un valor!')

if __name__ == '__main__':
        future = Future()
				future.add_done_callback(callback_futuro)
				future.add_done_callback(
						lambda result: logging.info('Definimos otro callback mas que sera ejecutado')
				)

				logging.info('''Hola, date cuenta que el futuro tiene callback, 
												pero no tiene un valor''')

				future.set_result('Soy un valor')

				logging.info('Ahora el futuro tiene un valor!!')

7:14