{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "

\n", "IIC2233 Apunte Programación Avanzada
\n", "© 2015 Karim Pichara - Christian Pieringer. Todos los derechos reservados.\n", "
\n", " Modificado desde 2017-2 al 2026-1 por Equipo Docente IIC2233\n", "

" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Tabla de contenidos\n", "\n", "1. [Sincronización](#Sincronización)\n", "2. [Mecanismos de sincronización](#Mecanismos-de-sincronización)\n", " 1. [*Lock*](#Lock)\n", " 2. [*Locks* y *Context Managers*](#Locks-y-Context-Managers)\n", " 3. [Señales entre *threads*: `Event`](#Señales-entre-threads:-Event)\n", " 4. [Otros métodos de coordinación entre *threads*](#Otros-métodos-de-coordinación-entre-threads)\n", "3. [*Deadlocks*](#Deadlocks)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Sincronización\n", "\n", "En el *notebook* anterior, vimos qué eran los *threads*, cómo crearlos, y hablamos sobre herramientas que podrían ser útiles. Podemos hacer cosas **pseudo-paralelas**, ¿qué podría salir mal?\n", "\n", "Hagamos dos *threads* que aumenten un mismo contador $10^6$ veces cada uno. Lo que esperaríamos es que el valor final sea $2 \\times 10^6$, ¿no es así?" ] }, { "cell_type": "code", "execution_count": 1, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Listo, nuestro contador vale 1279213\n" ] } ], "source": [ "import threading\n", "\n", "\n", "class Contador: \n", " def __init__(self) -> None:\n", " self.valor = 0\n", "\n", " \n", "def sumador(contador: int) -> None:\n", " for _ in range(10**6): ### Propuesto: cambiar este valor a 10**2. ¿Se comporta igual?\n", " contador.valor += int(1) # Este int() redundante es solo para que el ejemplo funcione\n", "\n", "\n", "contador = Contador() \n", "t1 = threading.Thread(target=sumador, args=(contador,))\n", "t2 = threading.Thread(target=sumador, args=(contador,))\n", "\n", "t1.start()\n", "t2.start()\n", "t1.join()\n", "t2.join()\n", "\n", "print(\"Listo, nuestro contador vale\", contador.valor)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "😱¿Qué pasó ahí? Como dice un viejo refrán:\n", "\n", "> Un programador tenía un problema y decidió usar *threads*. Ahora tiene dos problemas.\n", "\n", "En este ejemplo, intentamos modificar **al mismo tiempo (concurrentemente)** un mismo valor o recurso con dos *threads* distintos. Para entender por qué eso no siempre resulta como pensábamos, tomemos en cuenta que:\n", "\n", "- Las operaciones de los *threads* pueden ser pausadas en cualquier momento para dar paso a otro *thread*.\n", "- Es imposible saber cómo se entremezcla la ejecución de los *threads*. Es una decisión del sistema operativo. Por ejemplo, no podemos suponer que **siempre** se ejecutará una operación del *thread* 1, luego una del *thread* 2, una del *thread* 1, y así sucesivamente.\n", "\n", "Descompongamos en un pseudocódigo –lo más granular posible– lo que hace la máquina en una iteración en cada `sumador`:\n", "\n", " 1. Leer el valor de contador.valor\n", " 2. Sumar 1 al valor anterior\n", " 3. Almacenar el valor obtenido en contador.valor \n", " \n", "Ahora veamos un escenario posible:\n", "\n", " - T1 lee 0 de contador.valor\n", " - T1 suma 1 => 1\n", " - T1 guarda 1 en contador.valor\n", " - T1 lee 1 de contador.valor\n", " - T1 se pausa\n", " - T2 lee 1 de contador.valor\n", " - T2 suma 1 => 2\n", " - T2 guarda 2 en contador.valor\n", " - T2 lee 2 de contador.valor\n", " - T2 suma 1 => 3\n", " - T2 guarda 3 en contador.valor\n", " - T2 se pausa\n", " - T1 se reanuda\n", " - T1 suma 1 => 2 (😨)\n", " - T1 guarda 2 en contador.valor (😨😨😨)\n", " - ..." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Revisemos el caso mediante código:" ] }, { "cell_type": "code", "execution_count": 2, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "T1: lee 0\n", "T1: suma 1 => 1\n", "T1: guarda 1\n", "T2: lee 1\n", "T2: suma 1 => 2\n", "T2: guarda 2\n", "T1: lee 2T2: lee 2\n", "T2: suma 1 => 3\n", "T2: guarda 3\n", "\n", "T1: suma 1 => 3\n", "T1: guarda 3\n", "T2: lee 3T1: lee 3\n", "T1: suma 1 => 4\n", "T1: guarda 4\n", "\n", "T2: suma 1 => 4\n", "T2: guarda 4\n", "T1: lee 4T2: lee 4\n", "T2: suma 1 => 5\n", "T2: guarda 5\n", "\n", "T1: suma 1 => 5\n", "T1: guarda 5\n", "T1: lee 5T2: lee 5\n", "T2: suma 1 => 6\n", "T2: guarda 6\n", "\n", "T1: suma 1 => 6\n", "T1: guarda 6\n", "T2: lee 6T1: lee 6\n", "T1: suma 1 => 7\n", "T1: guarda 7\n", "\n", "T2: suma 1 => 7\n", "T2: guarda 7\n", "T1: lee 7T2: lee 7\n", "T2: suma 1 => 8\n", "T2: guarda 8\n", "\n", "T1: suma 1 => 8\n", "T1: guarda 8\n", "T1: lee 8T2: lee 8\n", "T2: suma 1 => 9\n", "T2: guarda 9\n", "\n", "T1: suma 1 => 9\n", "T1: guarda 9\n", "T2: lee 9T1: lee 9\n", "T1: suma 1 => 10\n", "T1: guarda 10\n", "\n", "T2: suma 1 => 10\n", "T2: guarda 10\n", "T2: lee 10T1: lee 10\n", "T1: suma 1 => 11\n", "T1: guarda 11\n", "\n", "T2: suma 1 => 11\n", "T2: guarda 11\n", "Listo, nuestro contador vale 11\n" ] } ], "source": [ "import time\n", "\n", "\n", "def sumador(contador: int) -> None:\n", " nombre = threading.current_thread().name\n", " for _ in range(10):\n", " valor = contador.valor\n", " print(f\"{nombre}: lee {valor}\")\n", " nuevo_valor = valor + 1\n", " print(f\"{nombre}: suma 1 => {nuevo_valor}\")\n", " contador.valor = nuevo_valor\n", " print(f\"{nombre}: guarda {nuevo_valor}\")\n", " time.sleep(1)\n", "\n", " \n", "contador = Contador() \n", "t1 = threading.Thread(name=\"T1\", target=sumador, args=(contador,))\n", "t2 = threading.Thread(name=\"T2\", target=sumador, args=(contador,))\n", "\n", "t1.start()\n", "t2.start()\n", "t1.join()\n", "t2.join()\n", "\n", "print(\"Listo, nuestro contador vale\", contador.valor)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "La situación anterior nos enseña que **deberíamos asegurarnos** de que la operación de aumentar el contador (`contador.valor += 1`) sea **atómica**, es decir, que solo un *thread* la pueda ejecutar en todo momento. Dentro de un código puede haber una secuencia de instrucciones que no queremos que sean ejecutadas en paralelo. Esto quiere decir solo un *thread* debe encontrarse en ese sección de código. A esta secuencia de instrucciones le llamamos **sección crítica**.\n", "\n", "La situación que vimos es una de muchas donde más de un *thread* debe compartir el acceso a determinados recursos, por ejemplo archivos, variables, etc. En estos escenarios, **solo uno** de los *threads* debe tener acceso al recurso y el resto debe quedar en espera para su uso. Cuando existe **concurrencia** múltiple a un recurso es posible controlar el acceso a este mediante mecanismos de **sincronización** entre los *threads*." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Mecanismos de sincronización\n", "\n", "Ahora, veremos dos formas de coordinar o **sincronizar** nuestros *threads*." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### *Lock*\n", "\n", "El *lock* es una primitiva de sincronización de *threads*, provista por la clase `Lock` de la librería `threading`. Se utiliza para que sólo un *thread* pueda estar en una misma sección crítica a la vez. En otras palabras, el *lock* permite la sincronización para el acceso a los recursos compartidos entre dos o más *threads*. \n", "\n", "Un *lock* puede estar **bloqueado** o **desbloqueado** (inicialmente está desbloqueado). Si un *thread* quiere entrar a una sección crítica, primero debe adquirir el *lock* mediante la operación `acquire()`. Una vez que el *thread* consigue adquirir el *lock*, lo deja bloqueado, haciendo que otros *threads* que quieran adquirir el mismo *lock* deben esperar. Cuando el *thread* quiera salir de la sección crítica, debe liberar el lock mediante `release()`, con lo que el *lock* queda desbloqueado, permitiendo que otro *thread* pueda adquirirlo.\n", "\n", "Una analogía a los *locks* es tener una zona protegida por llave. El primer *thread* en acceder a dicha zona, se lleva la llave y nadie puede entrar (el lock está **bloqueado**) y solo cuando este salga de dicha zona, devuelve la llave (lock **desbloqueado**) permitiendo que otro *thread* acceda a dicha zona.\n", "\n", "![lock](imgs/lock.png)\n", "\n", "Abajo, se ejemplifica el modo de usar un *lock*." ] }, { "cell_type": "code", "execution_count": 3, "metadata": {}, "outputs": [], "source": [ "import threading\n", "\n", "## El mismo lock para todos los threads\n", "lock_global = threading.Lock()\n", "\n", "\n", "def sumador_con_seccion_critica(contador: int, lock: threading.Lock) -> None:\n", " for _ in range(10 ** 6):\n", " # Pedimos el lock antes de entrar a la sección crítica.\n", " lock.acquire()\n", " # --- Sección crítica ---. \n", " # Está garantizado que en estas líneas sólo habrá un thread a la vez.\n", " contador.valor += int(1)\n", " # --- Fin de la sección crítica ---.\n", " # Liberamos el lock luego de salir de la sección crítica.\n", " lock.release()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Nuestro ejemplo inicial del contador ahora funcionará correctamente 👍. " ] }, { "cell_type": "code", "execution_count": 4, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Listo, nuestro contador vale 2000000\n" ] } ], "source": [ "contador = Contador() \n", "t1 = threading.Thread(target=sumador_con_seccion_critica, args=(contador,lock_global))\n", "t2 = threading.Thread(target=sumador_con_seccion_critica, args=(contador,lock_global))\n", "\n", "# Esto podría tomar algunos segundos...\n", "t1.start()\n", "t2.start()\n", "t1.join()\n", "t2.join()\n", "\n", "print(\"Listo, nuestro contador vale\", contador.valor)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "#### ATENCIÓN: el *lock* debe ser compartido\n", "\n", "El *lock* que usan los *threads* **debe ser el mismo**. Si cada *thread* usa una instancia distinta de *lock*, todos podrán pasar el método `acquire()` y entrarán a la sección crítica sin problemas." ] }, { "cell_type": "code", "execution_count": 5, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Listo, nuestro contador vale 1572749\n" ] } ], "source": [ "import threading\n", "\n", "## Si cada thread usa su propio lock, no estamos protegiendonos\n", "## de ningún acceso concurrente.\n", "\n", "\n", "def sumador_con_seccion_critica_mala(contador: int) -> None:\n", " ## Cada thread crea su propio lock\n", " ## 🙅‍♂️🙅‍♀️ ESTO ESTÁ MAL 🙅‍♂️🙅‍♀️\n", " lock = threading.Lock()\n", "\n", " for _ in range(10 ** 6):\n", " # Pedimos el lock antes de entrar a la sección crítica.\n", " lock.acquire()\n", " # --- Sección crítica ---. \n", " contador.valor += int(1)\n", " # --- Fin de la sección crítica ---.\n", " # Liberamos el lock luego de salir de la sección crítica.\n", " lock.release()\n", " \n", " \n", "contador = Contador() \n", "t1 = threading.Thread(target=sumador_con_seccion_critica_mala, args=(contador,))\n", "t2 = threading.Thread(target=sumador_con_seccion_critica_mala, args=(contador,))\n", "\n", "t1.start()\n", "t2.start()\n", "\n", "t1.join()\n", "t2.join()\n", "\n", "print(\"Listo, nuestro contador vale\", contador.valor)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### *Locks* y *Context Managers*\n", "\n", "En Python los *locks* también pueden funcionar dentro de un *context manager* a través de la sentencia `with`. En este caso es el mismo `with` el que se encarga de llamar los métodos `acquire()` y `release()` de forma implícita. De esta forma el *lock* usado en la función `sumador_con_seccion_critica` quedaría como se indica a continuación:" ] }, { "cell_type": "code", "execution_count": 7, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "T1: lee 0\n", "T1: suma 1 => 1\n", "T1: guarda 1\n", "T1: lee 1\n", "T1: suma 1 => 2\n", "T1: guarda 2\n", "T2: lee 2\n", "T2: suma 1 => 3\n", "T2: guarda 3\n", "T2: lee 3\n", "T2: suma 1 => 4\n", "T2: guarda 4\n", "T2: lee 4\n", "T2: suma 1 => 5\n", "T2: guarda 5\n", "T1: lee 5\n", "T1: suma 1 => 6\n", "T1: guarda 6\n", "T1: lee 6\n", "T1: suma 1 => 7\n", "T1: guarda 7\n", "T1: lee 7\n", "T1: suma 1 => 8\n", "T1: guarda 8\n", "T1: lee 8\n", "T1: suma 1 => 9\n", "T1: guarda 9\n", "T2: lee 9\n", "T2: suma 1 => 10\n", "T2: guarda 10\n", "T2: lee 10\n", "T2: suma 1 => 11\n", "T2: guarda 11\n", "T1: lee 11\n", "T1: suma 1 => 12\n", "T1: guarda 12\n", "T1: lee 12\n", "T1: suma 1 => 13\n", "T1: guarda 13\n", "T1: lee 13\n", "T1: suma 1 => 14\n", "T1: guarda 14\n", "T2: lee 14\n", "T2: suma 1 => 15\n", "T2: guarda 15\n", "T2: lee 15\n", "T2: suma 1 => 16\n", "T2: guarda 16\n", "T1: lee 16\n", "T1: suma 1 => 17\n", "T1: guarda 17\n", "T2: lee 17\n", "T2: suma 1 => 18\n", "T2: guarda 18\n", "T2: lee 18\n", "T2: suma 1 => 19\n", "T2: guarda 19\n", "T2: lee 19\n", "T2: suma 1 => 20\n", "T2: guarda 20\n" ] } ], "source": [ "import time\n", "\n", "\n", "lock_global = threading.Lock()\n", "\n", "\n", "def sumador(contador: int, lock: threading.Lock) -> None:\n", " nombre = threading.current_thread().name\n", " for _ in range(10):\n", " with lock:\n", " # --- Sección crítica ---. \n", " # Está garantizado que en estas líneas sólo habrá un thread a la vez.\n", " valor = contador.valor\n", " print(f\"{nombre}: lee {valor}\")\n", " nuevo_valor = valor + 1\n", " print(f\"{nombre}: suma 1 => {nuevo_valor}\")\n", " contador.valor = nuevo_valor\n", " print(f\"{nombre}: guarda {nuevo_valor}\")\n", " time.sleep(1)\n", " # --- Fin de la sección crítica ---.\n", "\n", " \n", "contador = Contador() \n", "t1 = threading.Thread(name=\"T1\", target=sumador, args=(contador,lock_global))\n", "t2 = threading.Thread(name=\"T2\", target=sumador, args=(contador,lock_global))\n", "\n", "t1.start()\n", "t2.start()\n", "\n", "t1.join()\n", "t2.join()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Señales entre *threads*: `Event`\n", "\n", "Otro patrón común que aparece en coordinación entre *threads* es la comunicación de eventos. Esto se refiere a la idea de que un *thread* sea capaz de esperar a que cierto evento ocurra para continuar sus operaciones. Por ejemplo, un *thread* que debe esperar a que ciertos cómputos se hayan completado previamente para poder operar con esos resultados. El método `join` podría ayudar a este cometido. El *thread* que espera puede usar `join` del (de los) *thread(s)* que espera. Pero `join` solo funciona para esperar el **término** de un *thread*, pero\n", "* ¿Qué ocurre si se desea esperar un evento que ocurre **durante** la vida de otro *thread*? \n", "* ¿Podemos tener un comportamiento similar pero donde el segundo *thread* no tenga necesariamente que terminar para que el primer *thread* continúe? \n", "* ¿Cómo podemos hacer que un *thread* espere a que otro le avise cuando continuar?\n", "\n", "Para ello tenemos los objetos `Event`. Un *event* es uno de los mecanismos más simples de comunicación entre *threads*: un *thread* hace una señal, y otros *threads* esperan a que esa señal ocurra. Los `Event` tienen un *flag* interno, que toma el valor `True` cuando la señal está activa, y `False` cuando no.\n", "\n", "* **Esperar**: Un *thread* puede esperar una señal llamando al método `wait()` del `Event`; con ello, el *thread* quedará en pausa hasta que otro *thread* active la señal correspondiente. En caso de que la señal ya haya estado activa antes de hacer `wait()`, el *thread* puede seguir inmediatamente sin esperar.\n", "\n", "* **Avisar**: Para activar la señal, un *thread* debe llamar al método `set()`, que dejará el *flag* interno del objeto `Event` en `True`. Finalmente, un *thread* cualquiera puede **resetear** la señal llamando a `clear()` del objeto `Event`, dejando el *flag* en `False`.\n", "\n", "* **Revisar**: Para revisar si una señal fue activada o no, se debe llamar al método `is_set()` de un *thread*. Este retornará `True` o `False` indicando el valor del *flag* del *thread* sin tener que pausar el thread como ocurre cuando se usa `wait()`.\n", "\n", "En el siguiente ejemplo, los *thread 2 y 3* usan `E.wait()` para esperar que el evento `E` se active. Despues de un rato, el *thread 1* hace `E.set()` para que el evento `E` se active y así los *threads 2 y 3* pueden continuar con su ejecución.\n", "\n", "![](imgs/event.png)\n", "\n", "\n", "\n", "Un ejemplo es cuando queremos reproducir un audio y un video de la forma más sincronizada posible. Supongamos que tenemos un *thread* encargado de leer el audio, y otro de leer el video. El *thread* encargado del audio debería esperar a que el *thread* encargado del video esté listo para empezar, y viceversa:" ] }, { "cell_type": "code", "execution_count": 8, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "[11:58:55] Cargando audio Your Name - Zen Zen Zense. Se demorará 3 segundos\n", "[11:58:55] Cargando video Your Name - Zen Zen Zense. Se demorará 10 segundos\n", "\n", "\n", "[11:58:58] Audio cargado. Esperando video\n", "[11:59:05] Video cargado. Esperando audio\n", "[11:59:05] Comenzando reproducción del video\n", "[11:59:05] Comenzando reproducción del audio\n" ] } ], "source": [ "# Ejemplo adaptado de http://zulko.github.io/blog/2013/09/19/a-basic-example-of-threads-synchronization-in-python/\n", "\n", "import threading\n", "import time\n", "\n", "\n", "# Tenemos dos eventos o señales.\n", "# Esta es para avisar que el video ya está listo para ser reproducido.\n", "video_cargado = threading.Event()\n", "\n", "# Esta es para avisar que el audio ya está listo para ser reproducido.\n", "audio_cargado = threading.Event()\n", "\n", "def hora_actual() -> None:\n", " return time.ctime().split(\" \")[3]\n", "\n", " \n", "def reproducir_video(nombre: str) -> None:\n", " print(f\"[{hora_actual()}] Cargando video {nombre}. Se demorará 10 segundos\\n\")\n", " # Supongamos que se demora 10 segundos\n", " time.sleep(10)\n", " print(f\"[{hora_actual()}] Video cargado. Esperando audio\")\n", " # Avisamos que el video ya está cargado\n", " video_cargado.set()\n", " # Esperamos a que el audio ya se haya cargado\n", " audio_cargado.wait()\n", " # ¡Listo!\n", " print(f\"[{hora_actual()}] Comenzando reproducción del video\")\n", " \n", " \n", "def reproducir_audio(nombre: str) -> None:\n", " print(f\"[{hora_actual()}] Cargando audio {nombre}. Se demorará 3 segundos\\n\")\n", " # Supongamos que se demora 3 segundos\n", " time.sleep(3)\n", " print(f\"[{hora_actual()}] Audio cargado. Esperando video\")\n", " # Avisamos que el audio ya está cargado\n", " audio_cargado.set()\n", " # Esperamos a que el video ya se haya cargado\n", " video_cargado.wait()\n", " # ¡Listo!\n", " print(f\"[{hora_actual()}] Comenzando reproducción del audio\")\n", " \n", " \n", "t1 = threading.Thread(target=reproducir_audio, args=(\"Your Name - Zen Zen Zense\",))\n", "t2 = threading.Thread(target=reproducir_video, args=(\"Your Name - Zen Zen Zense\",))\n", "\n", "t1.start()\n", "t2.start()\n", "\n", "t1.join()\n", "t2.join()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "En el ejemplo anterior, gracias a la coordinación de *threads* con *events*, conseguimos que el audio y el video se empiecen a reproducir **casi** simultáneamente. De otra manera, el audio habría empezado mucho antes, puesto que demoró solo 3 segundos en cargar, mientras que el video tardó 10 segundos.\n", "\n", "El siguiente ejemplo hace uso se los métodos `set()`, `clear()`, `wait()`, `is_set()` de los *threads* en un mismo problema. Vamos a tener dos *threads*, los cuales simulan a un **trabajador** y a un **jefe** de un proyecto. El trabajador hará sus labores por una cantidad definida de semanas, y cada 4 semanas solicitará su sueldo, pausando sus labores hasta recibirlo. Por parte del jefe, este seguirá trabajando mientras el trabajador no haya terminado sus labores, además de periódicamente revisar si le han solicitado un sueldo o no. En caso de recibir la solicitud, este pagará al trabajador y esperará la confirmación de este, antes de seguir con sus labores. Una vez el trabajador terminé todas sus tareas, avisará y con eso tanto trabajador como jefe habrán finalizado el proyecto." ] }, { "cell_type": "code", "execution_count": 9, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "[Jefe] Tenemos que terminar este proyecto! 👷‍♀️\n", "[Jefe] Trabajando...\n", "[Jefe] Todavía no me piden el pago de este mes 🤔\n", "[Trabajador] A chambear se ha dicho 👷‍♂️\n", "[Trabajador] Trabajando la semana #1\n", "[Jefe] Trabajando...\n", "[Jefe] Todavía no me piden el pago de este mes 🤔\n", "[Trabajador] Trabajando la semana #2\n", "[Jefe] Trabajando...\n", "[Jefe] Todavía no me piden el pago de este mes 🤔\n", "[Trabajador] Trabajando la semana #3\n", "[Jefe] Trabajando...\n", "[Jefe] Todavía no me piden el pago de este mes 🤔\n", "[Trabajador] Trabajando la semana #4\n", "[Jefe] Trabajando...\n", "[Jefe] Todavía no me piden el pago de este mes 🤔\n", "[Trabajador] ¡Jefe! Ya estamos a fin de mes 🤑\n", "[Jefe] Trabajando...\n", "[Jefe] Aquí tiene su dinero estimado 💵\n", "[Trabajador] A seguir trabajando pues 💰\n", "[Trabajador] Trabajando la semana #5\n", "[Trabajador] Trabajando la semana #6[Jefe] Trabajando...\n", "[Jefe] Todavía no me piden el pago de este mes 🤔\n", "\n", "[Jefe] Trabajando...[Trabajador] Trabajando la semana #7\n", "\n", "[Jefe] Todavía no me piden el pago de este mes 🤔\n", "[Trabajador] Trabajando la semana #8[Jefe] Trabajando...\n", "[Jefe] Todavía no me piden el pago de este mes 🤔\n", "\n", "[Jefe] Trabajando...[Trabajador] ¡Jefe! Ya estamos a fin de mes 🤑\n", "\n", "[Jefe] Aquí tiene su dinero estimado 💵\n", "[Trabajador] A seguir trabajando pues 💰\n", "[Trabajador] Trabajando la semana #9\n", "[Trabajador] Trabajando la semana #10[Jefe] Trabajando...\n", "[Jefe] Todavía no me piden el pago de este mes 🤔\n", "\n", "[Jefe] Trabajando...[Trabajador] Trabajando la semana #11\n", "\n", "[Jefe] Todavía no me piden el pago de este mes 🤔\n", "[Trabajador] Trabajando la semana #12[Jefe] Trabajando...\n", "[Jefe] Todavía no me piden el pago de este mes 🤔\n", "\n", "[Jefe] Trabajando...[Trabajador] ¡Jefe! Ya estamos a fin de mes 🤑\n", "\n", "[Jefe] Aquí tiene su dinero estimado 💵\n", "[Trabajador] A seguir trabajando pues 💰\n", "[Trabajador] Terminé todas mis labores.\n", "[Jefe] Podemos dar el proyecto por finalizado 🎉\n" ] } ], "source": [ "import time\n", "import threading\n", "\n", "# Tenemos cuatro eventos o señales\n", "# Esta es para avisar que el trabajador solicitó su pago mensual\n", "solicitar_pago = threading.Event()\n", "\n", "# Esta es para avisar que el jefe pagó al trabajador\n", "entregar_pago = threading.Event()\n", "\n", "# Esta es para avisar que el trabajador recibió el pago\n", "confirmar_recepcion = threading.Event()\n", "\n", "# Esta le avisa al jefe que el trabajador terminó sus labores\n", "finalizar_trabajo = threading.Event()\n", "\n", "\n", "# El trabajador trabaja cada 4 semanas y espera su pago\n", "def trabajar_trabajador(semanas: int) -> None:\n", " print(\"[Trabajador] A chambear se ha dicho 👷‍♂️\")\n", "\n", " # Supongamos que se trabaja por semanas\n", " for semana in range(1, semanas + 1):\n", " print(f\"[Trabajador] Trabajando la semana #{semana}\")\n", " time.sleep(1)\n", "\n", " # Si la semana es múltiplo de 4, avisa que necesita pago\n", " # y deja de trabajar hasta que llegue el sueldo\n", " if not semana % 4:\n", " # Pide el pago e indica que no ha confirmado su recepción\n", " print(\"[Trabajador] ¡Jefe! Ya estamos a fin de mes 🤑\")\n", " confirmar_recepcion.clear()\n", " solicitar_pago.set()\n", "\n", " # Espera el pago\n", " entregar_pago.wait()\n", " print(\"[Trabajador] A seguir trabajando pues 💰\")\n", "\n", " # Avisa la recepción del pago\n", " confirmar_recepcion.set()\n", " # Apaga la señal de solicitar pago para esperar 4 semanas más\n", " solicitar_pago.clear()\n", "\n", " # Se acabó el proyecto a trabajar\n", " print(\"[Trabajador] Terminé todas mis labores.\")\n", " finalizar_trabajo.set()\n", "\n", "\n", "# El jefe deberá estar funcionando hasta que el proyecto termine\n", "# Y además, revisar que se le pague al trabajador cuando lo pida\n", "def trabajar_jefe() -> None:\n", " print(\"[Jefe] Tenemos que terminar este proyecto! 👷‍♀️\")\n", "\n", " finalizado = False\n", " while not finalizado:\n", " # Usa un is_set() para revisar si el trabajador ya terminó,\n", " # Pero no queremos detener el trabajo esperando el evento.\n", " # Es por esto que se usa is_set() en vez de wait() para revisar.\n", " if finalizar_trabajo.is_set():\n", " finalizado = True\n", " else:\n", " print(\"[Jefe] Trabajando...\")\n", "\n", " # El jefe en cada semana revisa si el trabajador ha solicitado sueldo.\n", " if solicitar_pago.is_set():\n", " # Le paga al trabajador\n", " print(\"[Jefe] Aquí tiene su dinero estimado 💵\")\n", " entregar_pago.set()\n", "\n", " # Ahora si, espera confirmación de recepción antes de seguir\n", " confirmar_recepcion.wait()\n", " # Apaga la señal de entregar pago hasta la siguiente solicitud\n", " entregar_pago.clear()\n", " else:\n", " print(f\"[Jefe] Todavía no me piden el pago de este mes 🤔\")\n", "\n", " # Un sleep para esperar un poco antes de volver a revisar\n", " time.sleep(1)\n", "\n", " # Se acabó el proyecto a trabajar\n", " print(\"[Jefe] Podemos dar el proyecto por finalizado 🎉\")\n", "\n", "\n", "# El proyecto de trabajo durará 12 semanas para el trabajador\n", "t_jefe = threading.Thread(target=trabajar_jefe)\n", "t_trabajador = threading.Thread(target=trabajar_trabajador, args=(12,))\n", "\n", "t_jefe.start()\n", "t_trabajador.start()\n", "\n", "t_jefe.join()\n", "t_trabajador.join()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Otros métodos de coordinación entre *threads*\n", "\n", "En Python existen otras maneras de coordinar *threads*, que son adecuadas para otras situaciones. No las veremos en detalle en este curso. Si gustas, puedes verlas en la [documentación](https://docs.python.org/3/library/threading.html#lock-objects), donde encontrarás información sobre objetos `RLock` (*lock* reentrante), objetos condición, objetos semáforo, objetos barrera y más." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## *Deadlocks*\n", "\n", "Introdujimos formas de coordinar *threads*, o de hacer que un *thread* espere al otro. Existen situaciones en que – por error – dos o más *threads* se esperan mutuamente, sin que ninguno finalmente avance. A este tipo de situaciones se le llama ***deadlock*** o **interbloqueo**, aunque hay nombres *menos afortunados*...\n", "\n", "![Abrazo mortal](imgs/abrazo-mortal.png)\n", "\n", "![Deadlock de tráfico](imgs/traffic-deadlock.jpg)\n", "\n", "\n", "Veamos dos ejemplos concretos de ***deadlocks***.\n", "\n", "Ejemplo con *locks*:" ] }, { "cell_type": "code", "execution_count": 10, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Worker: adquiriendo lock_2\n", "Master: adquiriendo lock_1\n", "Worker: adquiriendo lock_1\n", "Master: adquiriendo lock_2\n" ] } ], "source": [ "import threading\n", "import time\n", "\n", "\n", "lock_1 = threading.Lock()\n", "lock_2 = threading.Lock()\n", "\n", "\n", "def master() -> None:\n", " time.sleep(2)\n", " print(\"Master: adquiriendo lock_1\")\n", " with lock_1:\n", " time.sleep(2)\n", " print(\"Master: adquiriendo lock_2\")\n", " with lock_2:\n", " print(\"Master: ¡trabajando!\")\n", "\n", "\n", "def worker() -> None:\n", " time.sleep(1.5)\n", " print(\"Worker: adquiriendo lock_2\")\n", " with lock_2:\n", " time.sleep(2)\n", " print(\"Worker: adquiriendo lock_1\")\n", " with lock_1:\n", " print(\"Worker: ¡trabajando!\")\n", "\n", "\n", "t1 = threading.Thread(target=master)\n", "t2 = threading.Thread(target=worker)\n", "\n", "t1.start()\n", "t2.start()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "En el ejemplo anterior, tenemos dos *threads* y dos *locks*. El *thread master* alcanza a adquirir el `lock_1`, y el *thread worker* alcanza a adquirir el `lock_2`. Luego, *master* trata de adquirir `lock_2`, por lo que debe esperar que *worker* lo libere. Sin embargo, *worker* no liberará el `lock_2` sin antes poder adquirir `lock_1` 💀. Es por esto que ni *master* ni *worker* nos dicen que están trabajando.\n", "\n", "Ejemplo con *events*. Nota que es el mismo que el de la sección de [señales entre _threads_](#Señales-entre-threads), solo que cambiamos el orden en que se revisan/levantan las señales:" ] }, { "cell_type": "code", "execution_count": 11, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "[12:00:15] Cargando audio Your Name - Zen Zen Zense. Se demorará 3 segundos\n", "\n", "[12:00:15] Cargando video Your Name - Zen Zen Zense. Se demorará 10 segundos\n", "\n", "[12:00:18] Audio cargado. Esperando video\n", "[12:00:25] Video cargado. Esperando audio\n" ] } ], "source": [ "# Ejemplo adaptado de http://zulko.github.io/blog/2013/09/19/a-basic-example-of-threads-synchronization-in-python/\n", "\n", "import threading\n", "import time\n", "\n", "\n", "# Tenemos dos eventos o señales.\n", "# Esta es para avisar que el video ya está listo para ser reproducido.\n", "video_cargado = threading.Event()\n", "\n", "# Esta es para avisar que el audio ya está listo para ser reproducido.\n", "audio_cargado = threading.Event()\n", "\n", "\n", "def hora_actual():\n", " return time.ctime().split(\" \")[3]\n", "\n", "\n", "def reproducir_video(nombre: str) -> None:\n", " print(f\"[{hora_actual()}] Cargando video {nombre}. Se demorará 10 segundos\\n\")\n", " time.sleep(10)\n", " print(f\"[{hora_actual()}] Video cargado. Esperando audio\")\n", " audio_cargado.wait() # Esperar al audio\n", " video_cargado.set() # Avisar que el video está listo\n", "\n", " print(f\"[{hora_actual()}] Comenzando reproducción del video\")\n", "\n", "\n", "def reproducir_audio(nombre: str) -> None:\n", " print(f\"[{hora_actual()}] Cargando audio {nombre}. Se demorará 3 segundos\\n\")\n", " time.sleep(3)\n", " print(f\"[{hora_actual()}] Audio cargado. Esperando video\")\n", " video_cargado.wait() # Esperar al video\n", " audio_cargado.set() # Avisar que el audio está listo\n", "\n", " print(f\"[{hora_actual()}] Comenzando reproducción del audio\")\n", " \n", " \n", "t1 = threading.Thread(target=reproducir_audio, args=(\"Your Name - Zen Zen Zense\",))\n", "t2 = threading.Thread(target=reproducir_video, args=(\"Your Name - Zen Zen Zense\",))\n", "\n", "t1.start()\n", "t2.start()\n", "\n", "# ❗️❗️❗️❗️❗️❗️\n", "# Si descomentan las siguientes 2 líneas, la celda nunca se terminará de ejecutar.\n", "# Puesto que esperará que t1 termine. Situación que no ocurrirá.\n", "# ❗️❗️❗️❗️❗️❗️\n", "\n", "# t1.join() \n", "# t2.join()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Esencialmente, en el ejemplo anterior estamos esperando que el otro *thread* avise que hizo su trabajo, antes de avisar que el *thread* actual hizo el suyo. Esto produce que `reproducir_video` espere a que la señal de `audio_cargado` se active, antes de activar `video_cargado`. Sin embargo, `reproducir_audio` está esperando que `video_cargado` se active para luego activar `audio_cargado` 💀. La que puede, puede, pero estos *threads* no pueden." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Lo importante – para ti como programador o programadora – es saber que los ***deadlocks*** pueden ocurrir, y que debes tener cuidado al programar para que esto no te pase." ] } ], "metadata": { "kernelspec": { "display_name": "Python 3 (ipykernel)", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.12.10" }, "varInspector": { "cols": { "lenName": 16, "lenType": 16, "lenVar": 40 }, "kernels_config": { "python": { "delete_cmd_postfix": "", "delete_cmd_prefix": "del ", "library": "var_list.py", "varRefreshCmd": "print(var_dic_list())" }, "r": { "delete_cmd_postfix": ") ", "delete_cmd_prefix": "rm(", "library": "var_list.r", "varRefreshCmd": "cat(var_dic_list()) " } }, "types_to_exclude": [ "module", "function", "builtin_function_or_method", "instance", "_Feature" ], "window_display": false } }, "nbformat": 4, "nbformat_minor": 4 }