prefetch_multiplier_tuning

Ajuste fino del prefetch multiplier de los workers Celery para equilibrar throughput y latencia segun el tipo de tarea del pipeline de verificacion de identidad. Las tareas GPU-bound (face_match, liveness) requieren configuracion diferente a las CPU-bound (OCR, doc_processing) para evitar cuellos de botella y maximizar la utilizacion de recursos.

When to use

Usa esta skill cuando trabajes con el worker_pool_agent y necesites optimizar el rendimiento de los workers Celery en el pipeline KYC. Aplica cuando observes alta latencia en tareas GPU-bound, subutilizacion de workers CPU-bound, o desbalance de carga entre colas de verificacion.

Instructions

Identificar y clasificar las tareas del pipeline por tipo de recurso consumido:

# backend/modules/worker_pool/task_classification.py
TASK_RESOURCE_MAP = {
    "face_match": "gpu_bound",        # ArcFace inference
    "liveness_detection": "gpu_bound", # Anti-spoofing model
    "ocr_extraction": "cpu_bound",     # PaddleOCR / EasyOCR
    "doc_processing": "cpu_bound",     # OpenCV processing
    "antifraud_analysis": "cpu_bound", # ELA, metadata analysis
    "decision_engine": "io_bound",     # DB queries, scoring
}

Configurar prefetch multiplier diferenciado por tipo de worker:

# backend/modules/worker_pool/celery_config.py
# GPU-bound workers: prefetch=1 para evitar acaparar tareas mientras GPU esta ocupada
GPU_WORKER_CONFIG = {
    "worker_prefetch_multiplier": 1,
    "task_acks_late": True,
    "worker_concurrency": 1,  # Una tarea GPU a la vez
}

# CPU-bound workers: prefetch=4 para mantener pipeline lleno
CPU_WORKER_CONFIG = {
    "worker_prefetch_multiplier": 4,
    "task_acks_late": True,
    "worker_concurrency": 4,  # Multiples tareas CPU en paralelo
}

# IO-bound workers: prefetch=8 para alto throughput
IO_WORKER_CONFIG = {
    "worker_prefetch_multiplier": 8,
    "task_acks_late": False,
    "worker_concurrency": 8,
}

Crear scripts de lanzamiento de workers con configuracion especifica por cola:

# GPU workers - face_match y liveness
celery -A worker_pool worker \
    --queues=face_match,liveness \
    --concurrency=1 \
    --prefetch-multiplier=1 \
    --hostname=gpu-worker@%h

# CPU workers - OCR y document processing
celery -A worker_pool worker \
    --queues=ocr,doc_processing,antifraud \
    --concurrency=4 \
    --prefetch-multiplier=4 \
    --hostname=cpu-worker@%h

Implementar metricas para medir el impacto del prefetch multiplier:

# backend/modules/worker_pool/metrics.py
from prometheus_client import Histogram, Gauge

task_wait_time = Histogram(
    "kyc_task_wait_seconds",
    "Tiempo de espera en cola antes de ejecucion",
    ["task_name", "worker_type"],
)
prefetch_utilization = Gauge(
    "kyc_prefetch_utilization",
    "Ratio de tareas prefetched vs ejecutadas",
    ["worker_type"],
)

Implementar ajuste dinamico del prefetch multiplier basado en carga:

# backend/modules/worker_pool/dynamic_prefetch.py
from celery.signals import worker_ready
import psutil

@worker_ready.connect
def adjust_prefetch(sender, **kwargs):
    cpu_percent = psutil.cpu_percent(interval=1)
    current_multiplier = sender.app.conf.worker_prefetch_multiplier

    if cpu_percent > 85 and current_multiplier > 1:
        sender.app.conf.worker_prefetch_multiplier = max(1, current_multiplier - 1)
    elif cpu_percent < 40 and current_multiplier < 8:
        sender.app.conf.worker_prefetch_multiplier = current_multiplier + 1

Configurar task_acks_late junto con el prefetch para garantizar que tareas KYC no se pierdan:

# Combinacion critica para tareas de verificacion
app.conf.update(
    task_acks_late=True,           # ACK solo despues de completar
    task_reject_on_worker_lost=True,  # Reencolar si worker muere
    worker_prefetch_multiplier=1,  # Default conservador
)

Crear tests de carga para validar la configuracion de prefetch:

# backend/tests/test_prefetch_tuning.py
def test_gpu_worker_no_starvation():
    """Verificar que GPU workers no acaparan tareas."""
    results = []
    for _ in range(10):
        r = face_match_task.delay(test_session_id)
        results.append(r)
    wait_times = [r.result["wait_time"] for r in results]
    assert max(wait_times) < 8.0  # Dentro del SLA de 8 segundos

Notes

Un prefetch_multiplier=1 es obligatorio para tareas GPU-bound del pipeline KYC (face_match, liveness); valores mayores causan que un worker acapare tareas mientras la GPU esta ocupada, aumentando la latencia global.
Siempre combinar prefetch_multiplier con task_acks_late=True para tareas criticas de verificacion; esto evita perder tareas si un worker muere durante el procesamiento biometrico.
Revisar periodicamente las metricas de wait_time por cola para detectar desajustes; el objetivo es mantener el tiempo total de verificacion por debajo de 8 segundos segun los SLA del sistema.