Loading repository data…

G3-Critical-Sections-Without-Bottlenecks-Lab-Python GitHub Details, Stars and Alternatives | OpenRepoFinder

jerry-felipe / repository

G3-Critical-Sections-Without-Bottlenecks-Lab-Python

Laboratorio práctico en Python 3.13 para proteger secciones críticas sin convertir la aplicación en un cuello de botella, utilizando threading.Lock, RLock, semáforos, colas seguras y medición de concurrencia.

#python3

View Repository on GitHub ↗

REPOSITORY OVERVIEW

Live repository statistics

★ 1Stars

⑂ 0Forks

◯ 0Open issues

◉ 1Watchers

62/100

OPENREPOHUB HEALTH SIGNAL

Mixed signals

A transparent discovery signal based on current public GitHub metadata.

Recent activity35% weight

100

Community adoption25% weight

Maintenance state20% weight

100

License clarity10% weight

Project information10% weight

This score does not audit code, security, maintainers, documentation quality, or suitability. Verify the repository and its current documentation before adoption.

README preview

Python Critical Sections Without Bottlenecks

Laboratorio práctico para aprender a proteger recursos compartidos sin convertir toda la aplicación en una fila de espera.

El proyecto compara diferentes estrategias de sincronización en Python 3.13, mostrando cómo el tamaño y la ubicación de una sección crítica afectan la consistencia, la latencia, el rendimiento y la escalabilidad.

Idea clave: un buen lock protege únicamente el estado compartido; un mal lock bloquea trabajo que podría ejecutarse de manera concurrente.

El problema

Cuando varios hilos consultan o modifican simultáneamente un recurso compartido, pueden producirse:

Condiciones de carrera.
Actualizaciones perdidas.
Datos inconsistentes.
Inventarios negativos.
Resultados impredecibles.
Errores difíciles de reproducir.

La solución habitual consiste en utilizar sincronización o exclusión mutua. Sin embargo, proteger demasiado código también puede provocar:

Esperas innecesarias.
Alta contención.
Mayor latencia.
Menor throughput.
Baja escalabilidad.
Hilos bloqueados por operaciones que no necesitan protección.

El verdadero reto no es simplemente agregar un Lock, sino colocarlo alrededor de las instrucciones exactas que deben ejecutarse de forma indivisible.

Objetivo

Demostrar mediante código, pruebas y mediciones cómo:

Identificar el estado realmente compartido.
Delimitar correctamente una sección crítica.
Reducir el tiempo durante el cual se mantiene un lock.
Ejecutar operaciones independientes fuera del bloqueo.
Comparar diferentes mecanismos de sincronización.
Detectar condiciones de carrera.
Medir latencia, throughput y contención.

Caso de estudio

El proyecto simula un sistema concurrente de reservas de inventario.

Varios hilos intentan reservar unidades de un producto al mismo tiempo. El sistema debe garantizar que:

El inventario nunca sea negativo.
No se vendan más unidades de las disponibles.
Cada actualización sea consistente.
Las operaciones independientes no sean bloqueadas.
El sistema mantenga un buen nivel de concurrencia.

Estrategias implementadas

from __future__ import annotations

import threading
import time


class CoarseGrainedInventoryService:
    def __init__(self, initial_stock: int) -> None:
        self._stock = initial_stock
        self._lock = threading.Lock()

    def reserve(self, product_id: str, quantity: int) -> bool:
        with self._lock:
            self._validate_request(product_id, quantity)

            # Estas operaciones no deberían bloquear a otros hilos.
            self._simulate_external_validation()
            self._simulate_database_query()

            if self._stock < quantity:
                return False

            self._stock -= quantity

            # Tampoco deberían ejecutarse dentro del lock.
            self._simulate_audit()
            self._simulate_notification()

            return True

    @staticmethod
    def _validate_request(product_id: str, quantity: int) -> None:
        if not product_id.strip():
            raise ValueError("product_id is required")

        if quantity <= 0:
            raise ValueError("quantity must be greater than zero")

    @staticmethod
    def _simulate_external_validation() -> None:
        time.sleep(0.01)

    @staticmethod
    def _simulate_database_query() -> None:
        time.sleep(0.01)

    @staticmethod
    def _simulate_audit() -> None:
        time.sleep(0.005)

    @staticmethod
    def _simulate_notification() -> None:
        time.sleep(0.005)

from __future__ import annotations

import threading
import time
from dataclasses import dataclass
from enum import StrEnum


class ReservationStatus(StrEnum):
    APPROVED = "approved"
    REJECTED = "rejected"


@dataclass(frozen=True, slots=True)
class ReservationResult:
    product_id: str
    requested_quantity: int
    remaining_stock: int
    status: ReservationStatus


@dataclass(frozen=True, slots=True)
class ReservationContext:
    product_id: str
    quantity: int


class OptimizedInventoryService:
    def __init__(self, initial_stock: int) -> None:
        if initial_stock < 0:
            raise ValueError("initial_stock cannot be negative")

        self._available_stock = initial_stock
        self._inventory_lock = threading.Lock()

    def reserve(
        self,
        product_id: str,
        quantity: int,
    ) -> ReservationResult:
        self._validate_request(product_id, quantity)

        # Trabajo independiente realizado fuera del lock.
        context = self._prepare_reservation(product_id, quantity)

        with self._inventory_lock:
            # Solo el acceso al estado compartido forma parte
            # de la sección crítica.
            if self._available_stock < quantity:
                result = ReservationResult(
                    product_id=product_id,
                    requested_quantity=quantity,
                    remaining_stock=self._available_stock,
                    status=ReservationStatus.REJECTED,
                )
            else:
                self._available_stock -= quantity

                result = ReservationResult(
                    product_id=product_id,
                    requested_quantity=quantity,
                    remaining_stock=self._available_stock,
                    status=ReservationStatus.APPROVED,
                )

        # Auditoría, métricas y notificaciones fuera del lock.
        self._complete_reservation(context, result)

        return result

    def get_available_stock(self) -> int:
        with self._inventory_lock:
            return self._available_stock

    @staticmethod
    def _validate_request(product_id: str, quantity: int) -> None:
        if not product_id or not product_id.strip():
            raise ValueError("product_id is required")

        if quantity <= 0:
            raise ValueError("quantity must be greater than zero")

    @staticmethod
    def _prepare_reservation(
        product_id: str,
        quantity: int,
    ) -> ReservationContext:
        return ReservationContext(
            product_id=product_id,
            quantity=quantity,
        )

    @staticmethod
    def _complete_reservation(
        context: ReservationContext,
        result: ReservationResult,
    ) -> None:
        del context, result

        # Simulación de auditoría, métricas o notificaciones.
        time.sleep(0.001)

with self._inventory_lock:
    if self._available_stock >= quantity:
        self._available_stock -= quantity

Validar solicitud
       │
       ▼
Preparar operación
       │
       ▼
Solicitar el lock
       │
       ▼
┌─────────────────────────────┐
│      SECCIÓN CRÍTICA        │
│                             │
│  Consultar inventario       │
│  Actualizar inventario      │
│  Construir resultado        │
└─────────────────────────────┘
       │
       ▼
Liberar el lock
       │
       ▼
Auditoría, métricas y eventos

from __future__ import annotations

import threading


class InventoryBusyError(RuntimeError):
    pass


class TimeoutInventoryService:
    def __init__(self, initial_stock: int) -> None:
        self._stock = initial_stock
        self._lock = threading.Lock()

    def reserve(
        self,
        quantity: int,
        timeout_seconds: float = 0.5,
    ) -> bool:
        acquired = self._lock.acquire(timeout=timeout_seconds)

        if not acquired:
            raise InventoryBusyError(
                "The inventory resource is temporarily busy"
            )

        try:
            if self._stock < quantity:
                return False

            self._stock -= quantity
            return True
        finally:
            self._lock.release()

with lock:
    update_shared_state()

from __future__ import annotations

import threading
from collections import defaultdict


class ProductInventoryService:
    def __init__(self, initial_inventory: dict[str, int]) -> None:
        self._inventory = initial_inventory.copy()
        self._locks: defaultdict[str, threading.Lock] = defaultdict(
            threading.Lock
        )

    def reserve(self, product_id: str, quantity: int) -> bool:
        if quantity <= 0:
            raise ValueError("quantity must be greater than zero")

        product_lock = self._locks[product_id]

        with product_lock:
            available = self._inventory.get(product_id, 0)

            if available < quantity:
                return False

            self._inventory[product_id] = available - quantity
            return True

Producto A ── Lock A
Producto B ── Lock B
Producto C ── Lock C

from __future__ import annotations

from concurrent.futures import ThreadPoolExecutor
from time import perf_counter

from critical_section.service.optimized_inventory_service import (
    OptimizedInventoryService,
)
from critical_section.model.reservation_result import ReservationStatus


def run_simulation() -> None:
    initial_stock = 10_000
    total_requests = 100_000
    workers = 100

    service = OptimizedInventoryService(initial_stock)

    started_at = perf_counter()

    with ThreadPoolExecutor(
        max_workers=workers,
        thread_name_prefix="reservation-worker",
    ) as executor:
        futures = [
            executor.submit(
                service.reserve,
                "PRODUCT-001",
                1,
            )
            for _ in range(total_requests)
        ]

        results = [future.result() for future in futures]

    elapsed_seconds = per

Sin sincronización	Modifica el inventario sin protección	Evidenciar condiciones de carrera
Bloqueo amplio	Protege el método completo	Mostrar el impacto de bloquear demasiado
Bloqueo reducido	Protege solo la actualización compartida	Reducir contención
`threading.Lock`	Exclusión mutua básica	Proteger una sección crítica
`threading.RLock`	Lock reentrante	Permitir adquisiciones anidadas
Lock con timeout	Evita esperas indefinidas	Controlar fallos por contención
Lock por producto	Cada recurso tiene su propio lock	Evitar un bloqueo global
`queue.Queue`	Comunicación segura entre hilos	Evitar manipulación manual compartida
`asyncio.Lock`	Exclusión mutua entre coroutines	Proteger estado en código asíncrono

G3-Critical-Sections-Without-Bottlenecks-Lab-Python

Live repository statistics

Mixed signals

README preview

Python Critical Sections Without Bottlenecks

El problema

Objetivo

Caso de estudio

Estrategias implementadas

Ejemplo de mala implementación

Implementación optimizada

Flujo de ejecución

Uso de timeout

Lock por producto

Simulación concurrente