{ "cells": [ { "cell_type": "markdown", "id": "350fb575", "metadata": {}, "source": [ "

\n", "IIC2233 Apunte Programación Avanzada
\n", " Creado el 2026-1 por Equipo Docente IIC2233. \n", "

" ] }, { "cell_type": "markdown", "id": "0f1aa21e", "metadata": {}, "source": [ "# Tabla de contenidos\n", "\n", "1. [Consultas concurrentes](#Consultas-concurrentes-y-Race-Conditions)\n", " 1. [Enviar múltiples requests en paralelo](#Enviar-múltiples-requests-en-paralelo)\n", " 2. [`ThreadPoolExecutor`: principales métodos](#ThreadPoolExecutor:-principales-métodos)\n", " 3. [Secuencial vs concurrente: comparación de tiempos](#Secuencial-vs-concurrente:-comparación-de-tiempos)\n", "\n", "2. [¿Qué es una *Race Condition*?](#¿Qué-es-una-Race-Condition?)\n", "\n", "3. [*Race Conditions* desde el cliente](#Race-Conditions-desde-el-cliente)\n", " 1. [Escenario 1: doble submit de formulario](#Escenario-1:-doble-submit-de-formulario)\n", " 2. [Escenario 2: *rate limiting* (código 429)](#Escenario-2:-rate-limiting-(código-429))\n", " 3. [Implementación: backoff con reintentos](#Implementación:-backoff-con-reintentos)\n", "\n", "4. [*Race Conditions* en servidores web](#Race-Conditions-en-servidores-web)\n", " 1. [El problema: estado compartido](#El-problema:-estado-compartido)\n", " 2. [Solución: usar un `Lock`](#Solución:-usar-un-Lock)\n" ] }, { "cell_type": "markdown", "id": "1e5c5e58", "metadata": {}, "source": [ "---\n", "\n", "## Consultas concurrentes y *Race Conditions*\n", "\n", "### Enviar múltiples requests en paralelo\n", "\n", "Hasta ahora hemos enviado las solicitudes **de forma secuencial**: esperamos la respuesta antes de enviar la siguiente. Pero a veces queremos enviar **muchas requests al mismo tiempo** para reducir el tiempo total de espera. Esto se puede hacer con `concurrent.futures.ThreadPoolExecutor`.\n", "\n", "```\n", "Secuencial: [req1]──►[req2]──►[req3]──► tiempo total = suma de tiempos individuales\n", "Concurrente: [req1]\n", " [req2] ←── todas en paralelo, tiempo total ≈ tiempo del más lento\n", " [req3]\n", "```" ] }, { "cell_type": "markdown", "id": "0bf816a8", "metadata": {}, "source": [ "### `ThreadPoolExecutor`: principales métodos\n", "\n", "`ThreadPoolExecutor` vive en el módulo `concurrent.futures` y gestiona un **pool de hilos** reutilizables. Sus métodos principales son:\n", "\n", "| Método | Descripción |\n", "|--------|-------------|\n", "| `ThreadPoolExecutor(max_workers=N)` | Crea el pool con hasta `N` hilos simultáneos. Si se omite, Python elige un valor por defecto basado en los núcleos del sistema. |\n", "| `executor.submit(fn, *args, **kwargs)` | Envía `fn(*args, **kwargs)` al pool y retorna un objeto `Future` **inmediatamente**, sin bloquear. |\n", "| `executor.map(fn, iterable)` | Versión simplificada: aplica `fn` a cada elemento del iterable y retorna los resultados **en orden de entrada**. Bloquea hasta que todos terminen. |\n", "| `future.result()` | Bloquea hasta obtener el valor de retorno del `Future`, o relanza la excepción si la función falló. |\n", "| `as_completed(futures)` | Generador que yield-ea cada `Future` a medida que **termina** (orden de completación, no de envío). |\n", "\n", "El patrón recomendado es usar el **context manager** (`with`): al salir del bloque, Python espera a que todos los hilos terminen (`shutdown(wait=True)`) antes de continuar.\n", "\n", "```python\n", "with ThreadPoolExecutor(max_workers=4) as executor:\n", " # submit → devuelve Future sin bloquear\n", " future = executor.submit(mi_funcion, arg1, arg2)\n", " resultado = future.result() # aquí sí bloquea\n", "\n", " # map → más simple, resultados en orden\n", " resultados = list(executor.map(mi_funcion, lista_de_args))\n", "\n", " # submit + as_completed → resultados en orden de llegada\n", " futuros = [executor.submit(mi_funcion, x) for x in lista_de_args]\n", " for f in as_completed(futuros):\n", " print(f.result())\n", "# al salir del 'with', se espera que todos los hilos terminen\n", "```" ] }, { "cell_type": "code", "execution_count": 4, "id": "1ff5f285", "metadata": {}, "outputs": [], "source": [ "import requests\n", "import time\n", "from concurrent.futures import ThreadPoolExecutor, as_completed\n", "\n", "BASE = \"https://dummyjson.com/\"\n", "\n", "\n", "def obtener_receta(receta_id: int) -> dict:\n", " \"\"\"Obtiene una receta individual por su id.\"\"\"\n", " respuesta = requests.get(f\"{BASE}recipes/{receta_id}\")\n", " return {\n", " \"id\": receta_id,\n", " \"status\": respuesta.status_code,\n", " \"nombre\": respuesta.json().get(\"name\"),\n", " }\n" ] }, { "cell_type": "code", "execution_count": 5, "id": "f181f2cd", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ " Receta 1 [200]: Classic Margherita Pizza\n", " Receta 2 [200]: Vegetarian Stir-Fry\n", " Receta 3 [200]: Chocolate Chip Cookies\n", " Receta 4 [200]: Chicken Alfredo Pasta\n", " Receta 5 [200]: Mango Salsa Chicken\n", " Receta 6 [200]: Quinoa Salad with Avocado\n", " Receta 7 [200]: Tomato Basil Bruschetta\n", " Receta 8 [200]: Beef and Broccoli Stir-Fry\n", " Receta 9 [200]: Caprese Salad\n", " Receta 10 [200]: Shrimp Scampi Pasta\n" ] } ], "source": [ "ids_recetas = list(range(1, 11)) # recetas 1 a 10\n", "\n", "with ThreadPoolExecutor(max_workers=10) as executor:\n", " futuros = {executor.submit(obtener_receta, i): i for i in ids_recetas}\n", " resultados_con = [futuro.result() for futuro in as_completed(futuros)]\n", "\n", "for r in sorted(resultados_con, key=lambda x: x[\"id\"]):\n", " print(f\" Receta {r['id']:2d} [{r['status']}]: {r['nombre']}\")\n" ] }, { "cell_type": "markdown", "id": "0f529ab1", "metadata": {}, "source": [ "### Secuencial vs concurrente: comparación de tiempos\n", "\n", "Ahora que conocemos los métodos principales de `ThreadPoolExecutor`, vamos a ver en la práctica la diferencia entre la forma **secuencial** y la forma **concurrente** de enviar requests, comparando el tiempo total de cada enfoque al obtener 10 recetas de la API de _DummyJSON_.\n" ] }, { "cell_type": "code", "execution_count": 6, "id": "827b9ef4", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Secuencial → 3.98s (10 recetas)\n", "Concurrente → 0.94s (10 recetas)\n", "Aceleración : 4.3x más rápido\n" ] } ], "source": [ "\n", "# --- Forma secuencial ---\n", "inicio = time.time()\n", "resultados_sec = [obtener_receta(i) for i in ids_recetas]\n", "tiempo_sec = time.time() - inicio\n", "print(f\"Secuencial → {tiempo_sec:.2f}s ({len(resultados_sec)} recetas)\")\n", "\n", "# --- Forma concurrente ---\n", "inicio = time.time()\n", "with ThreadPoolExecutor(max_workers=10) as executor:\n", " futuros = {executor.submit(obtener_receta, i): i for i in ids_recetas}\n", " resultados_con = [futuro.result() for futuro in as_completed(futuros)]\n", "tiempo_con = time.time() - inicio\n", "print(f\"Concurrente → {tiempo_con:.2f}s ({len(resultados_con)} recetas)\")\n", "print(f\"Aceleración : {tiempo_sec / tiempo_con:.1f}x más rápido\")\n" ] }, { "cell_type": "markdown", "id": "e7798839", "metadata": {}, "source": [ "---\n", "\n", "## ¿Qué es una *Race Condition*?\n", "\n", "Una ***race condition*** (condición de carrera) ocurre cuando **dos o más hilos o procesos acceden y modifican un recurso compartido de forma concurrente**, y el resultado final depende del orden arbitrario en que se ejecutan las operaciones. Como ese orden no está garantizado, el resultado puede ser **incorrecto e impredecible**.\n", "\n", "```\n", "Hilo A: lee valor = 5\n", "Hilo B: lee valor = 5 ← también leyó 5, no 6\n", "Hilo A: escribe valor = 6\n", "Hilo B: escribe valor = 6 ← ¡debería ser 7!\n", "```\n", "\n", "Este tipo de error —donde una escritura se pierde silenciosamente— se denomina ***lost update***. Las *race conditions* pueden aparecer:\n", "\n", "- **En el cliente:** cuando el usuario (o el código) envía requests duplicadas o simultáneas que afectan el mismo recurso.\n", "- **En el servidor:** cuando múltiples hilos atienden requests en paralelo y comparten estado mutable sin protección.\n", "\n", "Las secciones siguientes exploran ambos escenarios.\n" ] }, { "cell_type": "markdown", "id": "0fba77f8", "metadata": {}, "source": [ "---\n", "\n", "## *Race Conditions* desde el cliente\n", "\n", "Cuando el cliente envía múltiples requests concurrentes que **modifican el mismo recurso en el servidor**, puede provocar (o verse afectado por) una *race condition*. Algunos escenarios típicos:\n", "\n", "### Escenario 1: doble submit de formulario\n", "\n", "El usuario hace clic dos veces rápido en \"Comprar\". Se envían dos requests `POST` casi simultáneas. Si el servidor no tiene protección, el pedido puede duplicarse.\n", "\n", "**Mitigación en el cliente:** deshabilitar el botón/envío hasta recibir respuesta, o usar un token de idempotencia único por operación.\n", "\n", "```python\n", "import uuid\n", "\n", "token_idempotencia = str(uuid.uuid4()) # único por intento de compra\n", "\n", "respuesta = requests.post(\n", " \"https://api.tienda.com/compras\",\n", " json={\"producto_id\": 42, \"cantidad\": 1},\n", " headers={\"Idempotency-Key\": token_idempotencia}\n", ")\n", "```\n" ] }, { "cell_type": "markdown", "id": "1d2cc2e3", "metadata": {}, "source": [ "### Escenario 2: *rate limiting* (código 429)\n", "\n", "Enviar demasiadas requests en poco tiempo puede causar que el servidor rechace las siguientes con el código **`429 Too Many Requests`**. Esto no es una *race condition* en sí misma, pero es una consecuencia directa del uso concurrente de la API.\n", "\n", "**Mitigación:** respetar los límites de la API, agregar retardo entre intentos (*backoff* exponencial).\n" ] }, { "cell_type": "markdown", "id": "fdaa1e9e", "metadata": {}, "source": [ "### Implementación: backoff con reintentos\n", "\n", "La función `obtener_producto_con_backoff` reintenta la solicitud usando **backoff exponencial**: si recibe un `429`, espera $2^{\\text{intento}}$ segundos antes de volver a intentar ($1s$, $2s$, $4s$, …).\n" ] }, { "cell_type": "code", "execution_count": 7, "id": "1e35dedc", "metadata": {}, "outputs": [], "source": [ "def obtener_producto_con_backoff(producto_id: int, max_reintentos: int = 3) -> dict:\n", " \"\"\"\n", " Obtiene un producto y reintenta con espera exponencial si recibe 429 (Too Many Requests).\n", " \"\"\"\n", " for intento in range(max_reintentos):\n", " respuesta = requests.get(f\"{BASE}products/{producto_id}\")\n", "\n", " if respuesta.status_code == 429:\n", " # Backoff exponencial: intento=0 → 1s, intento=1 → 2s, intento=2 → 4s\n", " espera = 2 ** intento\n", " print(f\" [producto {producto_id}] 429 recibido, reintentando en {espera}s...\")\n", " time.sleep(espera)\n", " continue\n", "\n", " # Status distinto a 429: devolvemos el resultado (200, 404, etc.)\n", " return {\n", " \"id\": producto_id,\n", " \"status\": respuesta.status_code,\n", " \"titulo\": respuesta.json().get(\"title\"),\n", " }\n", "\n", " # Agotados todos los reintentos sin éxito\n", " return {\"id\": producto_id, \"status\": 429, \"titulo\": \"Sin respuesta tras reintentos\"}\n" ] }, { "cell_type": "markdown", "id": "e9467d97", "metadata": {}, "source": [ "#### Simulación con `unittest.mock`\n", "\n", "Como _DummyJSON_ no devuelve `429` fácilmente, usamos `unittest.mock.patch` para interceptar `requests.get` durante la prueba. Los productos **2 y 4** recibirán un `429` en su primer intento, y el producto **2** también en el segundo, forzando varios reintentos.\n" ] }, { "cell_type": "code", "execution_count": 8, "id": "d5ff1b12", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "=== Demostración con 429 simulado ===\n", "\n", " [producto 2] 429 recibido, reintentando en 1s...\n", " [producto 4] 429 recibido, reintentando en 1s...\n", " [producto 2] 429 recibido, reintentando en 2s...\n", "\n", " Producto 1 [200]: Producto simulado #1\n", " Producto 2 [200]: Producto simulado #2\n", " Producto 3 [200]: Producto simulado #3\n", " Producto 4 [200]: Producto simulado #4\n", " Producto 5 [200]: Producto simulado #5\n" ] } ], "source": [ "from unittest.mock import patch, MagicMock\n", "\n", "_llamadas = {} # cuenta cuántas veces se llamó por producto_id\n", "\n", "def mock_get(url, **kwargs):\n", " \"\"\"Simula que los productos 2 y 4 devuelven 429 en su primer intento.\"\"\"\n", " producto_id = int(url.split(\"/\")[-1])\n", " _llamadas[producto_id] = _llamadas.get(producto_id, 0) + 1\n", "\n", " # Productos 2 y 4: 429 en el primer intento\n", " if producto_id in (2, 4) and _llamadas[producto_id] == 1:\n", " resp = MagicMock()\n", " resp.status_code = 429\n", " return resp\n", "\n", " # Producto 2 también falla en el segundo intento\n", " if producto_id == 2 and _llamadas[producto_id] == 2:\n", " resp = MagicMock()\n", " resp.status_code = 429\n", " return resp\n", "\n", " # Resto: respuesta 200 exitosa simulada\n", " resp = MagicMock()\n", " resp.status_code = 200\n", " resp.json.return_value = {\"title\": f\"Producto simulado #{producto_id}\"}\n", " return resp\n", "\n", "\n", "print(\"=== Demostración con 429 simulado ===\\n\")\n", "with patch(\"requests.get\", side_effect=mock_get):\n", " with ThreadPoolExecutor(max_workers=5) as executor:\n", " futuros = [executor.submit(obtener_producto_con_backoff, i) for i in range(1, 6)]\n", " resultados = [f.result() for f in as_completed(futuros)]\n", "\n", "print()\n", "for r in sorted(resultados, key=lambda x: x[\"id\"]):\n", " print(f\" Producto {r['id']} [{r['status']}]: {r['titulo']}\")\n" ] }, { "cell_type": "markdown", "id": "49761774", "metadata": {}, "source": [ "---\n", "\n", "## *Race Conditions* en servidores web\n", "\n", "Flask puede atender **múltiples solicitudes en paralelo** (una por hilo con `threaded=True`). Si varios hilos leen y escriben el mismo dato sin coordinación, el resultado puede corromperse tal como se describió arriba.\n", "\n", "### El problema: estado compartido\n", "\n", "A continuación simulamos este problema con un **servidor Flask real**: enviamos 100 solicitudes `POST` concurrentes que incrementan un contador global. Sin protección, muchas escrituras se perderán.\n" ] }, { "cell_type": "code", "execution_count": 2, "id": "47b51b95", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Resultado esperado : 100\n", "Resultado obtenido : 56\n", "Valores perdidos : 44 ← race condition!\n" ] } ], "source": [ "import threading\n", "import time\n", "import logging\n", "import requests\n", "from werkzeug.serving import make_server\n", "from flask import Flask\n", "from concurrent.futures import ThreadPoolExecutor, as_completed\n", "\n", "logging.getLogger(\"werkzeug\").setLevel(logging.ERROR)\n", "\n", "# --- Servidor Flask SIN lock ---\n", "# threaded=True: cada request es atendida en su propio hilo → race condition real\n", "app_sin_lock = Flask(\"contador_sin_lock\")\n", "contador_sin_lock = 0\n", "\n", "@app_sin_lock.route(\"/incrementar\", methods=[\"POST\"])\n", "def incrementar_sin_lock():\n", " global contador_sin_lock\n", " valor = contador_sin_lock # leer\n", " time.sleep(0.001) # simula trabajo → amplifica la race condition\n", " contador_sin_lock = valor + 1 # escribir\n", " return {\"ok\": True}\n", "\n", "@app_sin_lock.route(\"/contador\", methods=[\"GET\"])\n", "def get_contador_sin_lock():\n", " return {\"contador\": contador_sin_lock}\n", "\n", "srv_sin_lock = make_server(\"localhost\", 4445, app_sin_lock, threaded=True)\n", "threading.Thread(target=srv_sin_lock.serve_forever, daemon=True).start()\n", "\n", "# Enviar 100 requests concurrentes al endpoint /incrementar\n", "N = 100\n", "with ThreadPoolExecutor(max_workers=N) as executor:\n", " futuros = [executor.submit(requests.post, \"http://localhost:4445/incrementar\")\n", " for _ in range(N)]\n", " for f in as_completed(futuros):\n", " f.result()\n", "\n", "resultado = requests.get(\"http://localhost:4445/contador\").json()[\"contador\"]\n", "srv_sin_lock.shutdown()\n", "\n", "print(f\"Resultado esperado : {N}\")\n", "print(f\"Resultado obtenido : {resultado}\")\n", "print(f\"Valores perdidos : {N - resultado} ← race condition!\")" ] }, { "cell_type": "markdown", "id": "a9361db2", "metadata": {}, "source": [ "### Solución: usar un `Lock`\n", "\n", "Un **`threading.Lock`** es un mecanismo de sincronización que garantiza que **solo un hilo a la vez** pueda ejecutar la sección crítica del código (la parte que accede al recurso compartido). Los demás hilos deben esperar hasta que el *lock* sea liberado.\n", "\n", "```\n", "Hilo A (request 1): adquiere lock → lee 5 → escribe 6 → libera lock\n", "Hilo B (request 2): espera... → → → adquiere lock → lee 6 → escribe 7 → libera lock\n", "```\n", "\n", "Ahora el resultado siempre es correcto.\n", "\n", "En Flask, el patrón es:\n", "```python\n", "from flask import Flask\n", "import threading\n", "\n", "app = Flask(__name__)\n", "lock = threading.Lock()\n", "contador = 0\n", "\n", "@app.route(\"/visitas\", methods=[\"GET\"])\n", "def visitas():\n", " global contador\n", " with lock: # solo un hilo a la vez entra aquí\n", " contador += 1\n", " return {\"visitas\": contador}\n", "```\n", "\n", "A continuación repetimos el mismo experimento con el `Lock` activado:\n" ] }, { "cell_type": "code", "execution_count": 10, "id": "11e846fe", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Resultado esperado : 100\n", "Resultado obtenido : 100\n", "Valores perdidos : 0 ← ninguno!\n" ] } ], "source": [ "# --- Servidor Flask CON lock ---\n", "app_con_lock = Flask(\"contador_con_lock\")\n", "_lock = threading.Lock()\n", "contador_con_lock = 0\n", "\n", "@app_con_lock.route(\"/incrementar\", methods=[\"POST\"])\n", "def incrementar_con_lock():\n", " global contador_con_lock\n", " with _lock: # sección crítica protegida\n", " valor = contador_con_lock\n", " time.sleep(0.001)\n", " contador_con_lock = valor + 1\n", " return {\"ok\": True}\n", "\n", "@app_con_lock.route(\"/contador\", methods=[\"GET\"])\n", "def get_contador_con_lock():\n", " return {\"contador\": contador_con_lock}\n", "\n", "srv_con_lock = make_server(\"localhost\", 4445, app_con_lock, threaded=True)\n", "threading.Thread(target=srv_con_lock.serve_forever, daemon=True).start()\n", "\n", "# Mismo experimento: 100 requests concurrentes\n", "N = 100\n", "with ThreadPoolExecutor(max_workers=N) as executor:\n", " futuros = [executor.submit(requests.post, \"http://localhost:4445/incrementar\")\n", " for _ in range(N)]\n", " for f in as_completed(futuros):\n", " f.result()\n", "\n", "resultado = requests.get(\"http://localhost:4445/contador\").json()[\"contador\"]\n", "srv_con_lock.shutdown()\n", "\n", "print(f\"Resultado esperado : {N}\")\n", "print(f\"Resultado obtenido : {resultado}\")\n", "print(f\"Valores perdidos : {N - resultado} ← ninguno!\")" ] } ], "metadata": { "kernelspec": { "display_name": ".venv", "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.3" } }, "nbformat": 4, "nbformat_minor": 5 }