Saltar al contenido
Ridgun
Español
EnglishEspañolPortuguêsSvenskaCatalàGalegoEuskaraItaliano

39 El Libro de Ridgun

La lib pool — Pool Acotado de Trabajadores, Round-Robin y Robo de Trabajo

39.1. ¿Por qué existe la lib pool? (El problema real y la filosofía Ridgun)

«¿Por qué necesito una lib pool escrita desde cero en Ridgun puro en lugar de enlazar una librería de C de toda la vida?»

Esa es la primera pregunta que se hace cualquier ingeniero de sistemas al llegar a pool. Y la respuesta resume toda la filosofía de ridgun-libs:

Seamos sinceros: lanzar un hilo o actor nuevo por cada petición entrante sin límite superior (unbounded spawning) hace que un pico de tráfico agote la memoria en segundos (Out Of Memory).

La lib pool proporciona un despachador de tareas con contrapresión explícita (PoolAccept, PoolRejectFull), partición equilibrada de lotes en $K$ carriles (pool partition), reequilibrio por robo de trabajo (work-stealing) (pool steal rebalance) y ejecución paralela sobre actores nativos.

En un ecosistema tradicional, resolver las tareas del dominio AGENT implica arrastrar dependencias externas en C, lidiar con punteros crudos, conversiones opacas de tipos, excepciones no verificadas o pausas impredecibles del recolector de basura. En Ridgun, la lib pool (src/main/ridgun/lib/pool.rg) aborda el problema con cuatro principios innegociables:

  1. Declara import pool (o import pool as alias) al inicio del archivo: cada lib define su espacio de nombres module :: pool y en modo --strict las funciones multi-palabra se invocan siempre con espacios naturales.
  2. Encadena tus tuberías de izquierda a derecha con ->: escribe entrada paso uno -> paso dos -> resultado, usa máscaras (+ _ _ +) para desestructurar secuencias y omite _ = en sentencias secuenciales (o descarta con -> ()).
  3. Recursión de cola con TCO (Tail-Call Optimization): todos los recorridos internos de pool se compilan a bucles nativos en $O(1)$ espacio de pila.
  4. Memoria determinista (RMM1 / RMM2): cada estructura intermedia se libera en el instante exacto en que termina su última lectura, sin Garbage Collector.

39.2. Artefactos nativos, importación y catálogo completo de la API de pool

En ridgun-libs, pool es una lib de primera clase ubicada en src/main/ridgun/lib/pool.rg (dominio AGENT). Al compilar con stable bundle, se empaqueta en tres artefactos nativos:

ArtefactoRutaPropósito
Bundle Ridguntarget/bundles/pool.rglibDistribución e importación directa con import pool
Archivo compiladotarget/pool.rgcArtefacto intermedio verificado por ridgunc
Objeto compartido ELF64target/libpool.soBiblioteca dinámica nativa sin dependencias de C ni libc

Cada lib declara su propio espacio de nombres con module :: pool: basta con escribir import pool (o import pool as alias) en el encabezado y llamar a sus funciones con espacios naturales (mi funcion x) y tuberías -> (x mi funcion -> resultado).

39.2.1. Resumen de familias funcionales de pool

Firma / Función / TipoQué hace y cuándo usarla
pool new : Int -> Int -> PoolStateCrea un pool acotado con workers carriles y capacidad máxima max_queue
pool submit : PoolState -> PoolTask -> (PoolState, PoolDecision)Encola una tarea en round-robin o devuelve PoolRejectFull si está saturado
pool partition / pool steal rebalanceReparte lotes entre $K$ trabajadores y roba la mitad de una cola sobrecargada para un trabajador libre
pool map sum / pool parallel sum2Despacha cargas de trabajo en paralelo sobre trabajadores agent nativos

39.2.2. Catálogo exhaustivo de tipos y firmas en src/main/ridgun/lib/pool.rg

A continuación tienes el inventario completo de tipos algebraicos (type), operadores (op ::) y firmas de funciones (fn ::) definidos en src/main/ridgun/lib/pool.rg (1 tipos, 0 operadores y 14 funciones):

Tipos algebraicos y registros (type) de pool:

type AgentPool = Pool Int [Agent]
#Función / Firma (fn ::)Tipo de Retorno
1fn :: pool size AgentPoolInt
2fn :: pool agents AgentPool[Agent]
3fn :: pool get agent [Agent] IntAgent
4fn :: pool abs IntInt
5fn :: pool ask shard AgentPool Int AB
6fn :: pool broadcast AgentPool A()
7fn :: pool broadcast list [Agent] A()
8fn :: pool drain ready Agent[B]
9fn :: pool drain ready loop Agent [B][B]
10fn :: pool drain step Agent Option [B][B]
11fn :: pool drain all AgentPool[B]
12fn :: pool drain list [Agent] [B][B]
13fn :: pool stop list [Agent]()
14fn :: pool shutdown AgentPool()

39.3. Explorando pool a fondo: Ejemplos paso a paso de cada funcionalidad

No basta con ver la tabla de firmas: vamos a explorar cada una de las funcionalidades de la lib pool mediante ejemplos progresivos y ejecutables en Ridgun. Empezaremos por las dos recetas tabulares de uso cotidiano y después recorreremos en detalle todos los escenarios funcionales, transformaciones y casos límite verificados en src/test/ridgun/test pool.rg.

39.3.1. Ejemplo 1: Flujo principal en estilo tabular con pool

Observa cómo en Ridgun los signos = y los pasos de transformación con pool quedan alineados como una tabla limpia, sin variables mutables innecesarias ni efectos laterales ocultos:

import pool

fn :: despachar a shard AgentPool Int String => String
      despachar a shard p hash_clave msg =
        pool ask shard p hash_clave msg

39.3.2. Ejemplo 2: Composición segura y manejo de casos de borde

Toda función de la lib pool está diseñada para componerse con guardas booleanas | y con el resto de las 32 libs de ridgun-libs:

import pool

fn :: difundir y cerrar AgentPool String => ()
      difundir y cerrar p evento =
        pool broadcast p evento
        pool shutdown p

39.3.3. Ejemplo 3: Combinadores de actores (agent ask, try ask, send batch, map batch, drain ready), pools particionados y reactor de conexiones

En este ejemplo práctico exploramos de forma exhaustiva cómo se comporta pool en este bloque funcional, incluyendo entradas estándar, combinaciones de parámetros y valores frontera:

fn :: test spec combinators pool and net agent => ()
      test spec combinators pool and net agent =
        w1 = agent (ag double worker)
        assert int equals (agent ask w1 21) 42 "agent ask 21 -> 42"
        mb = agent map batch w1 [3, 5, 7]
        assert int equals (ag sum ints mb) 30 "agent map batch [3,5,7] -> [6,10,14] sum 30"
        agent send batch w1 [10, 20] -> ()
        rc = agent receive count w1 2
        assert int equals (ag sum ints rc) 60 "agent send batch + agent receive count -> 60"
        c_dr = new counter agent 100
        send 5 to c_dr
        send 10 to c_dr
        dr = agent drain ready c_dr
        assert int equals (ag sum ints dr) 220 "agent drain ready [105, 115] -> 220"
        agent stop all [w1, c_dr] -> ()
        assert false (is agent alive w1) "w1 stopped by agent stop all"
        assert false (is agent alive c_dr) "c_dr stopped by agent stop all"

        s0 = new counter agent 0
        s1 = new counter agent 1000
        pl = Pool 2 [s0, s1]
        assert int equals (pool size pl) 2 "pool size 2"
        assert int equals (pool ask shard pl 0 5) 5 "pool ask shard 0 -> 5"
        assert int equals (pool ask shard pl 1 7) 1007 "pool ask shard 1 -> 1007"
        assert int equals (pool ask shard pl (0 - 3) 3) 1010 "pool ask shard negative key -> shard 1"
        pool broadcast pl 10 -> ()
        all_dr = pool drain all pl
        assert int equals (ag sum ints all_dr) 1035 "pool drain all [15, 1020] -> 1035"
        pool shutdown pl -> ()
        assert false (is agent alive s0) "pool s0 stopped"
        assert false (is agent alive s1) "pool s1 stopped"

        ca = agent (ag echo conn step) (CounterState 0)
        cb = agent (ag echo conn step) (CounterState 0)
        s_eof = conn feed payload (0 - 1) ca ""
        s_err = conn handle event (0 - 1) cb 8
        kept  = reactor sweep next s_eof (reactor sweep next s_err [])
        assert int equals (agent list len kept 0) 0 "reactor sweep next prunes closed slots"
        assert false (is agent alive ca) "ca stopped on EOF"
        assert false (is agent alive cb) "cb stopped on POLLERR"
        cc = agent (ag echo conn step) (CounterState 0)
        reactor shutdown [ConnSlot (0 - 1) cc] -> ()
        assert false (is agent alive cc) "cc stopped on reactor shutdown"

39.3.4. Ejemplo 4: Ciclo de vida de agentes (is agent alive, agent status, stop agent) y agente contador con estado aislado (new counter agent)

En este ejemplo práctico exploramos de forma exhaustiva cómo se comporta pool en este bloque funcional, incluyendo entradas estándar, combinaciones de parámetros y valores frontera:

fn :: test agent lifecycle and counter => ()
      test agent lifecycle and counter =
        assert str eq (format agent status AgentRunning) "running" "format AgentRunning"
        assert str eq (format agent status AgentStopped) "stopped" "format AgentStopped"
        assert str eq (format agent status (AgentTimedOut 250)) "timeout after 250 ms" "format AgentTimedOut"
        assert str eq (format agent status (AgentFailed "boom")) "failed: boom" "format AgentFailed"
        assert str eq (format agent reply (ReplyInt 42)) "42" "format ReplyInt"
        assert str eq (format agent reply (ReplyStr "ok")) "ok" "format ReplyStr"
        assert str eq (format agent reply (ReplyBool True)) "true" "format ReplyBool True"
        assert str eq (format agent reply (ReplyBool False)) "false" "format ReplyBool False"
        assert str eq (format agent reply ReplyEmpty) "agent outbox empty" "format ReplyEmpty"
        assert str eq (format agent reply ReplyStopped) "agent stopped" "format ReplyStopped"
        assert str eq (format agent reply (ReplyTimeout 50)) "agent timeout after 50 ms" "format ReplyTimeout"
        assert str eq (format agent reply (ReplyError "bad")) "agent error: bad" "format ReplyError"

        c   = new counter agent 10
        assert ok (is agent alive c)
        assert str eq (format agent status (agent status c)) "running" "counter running"
        assert str eq (format agent reply (try poll int c)) "agent outbox empty" "empty outbox before send"
        assert eq (counter get c) 10
        assert eq (counter add c 5) 15
        assert eq (counter add c 7) 22
        assert str eq (format agent reply (ask int c 8)) "30" "ask int 8 -> 30"
        assert str eq (format agent reply (ask int with timeout c 12 200)) "42" "ask int with timeout -> 42"
        assert eq (counter reset c) 0
        assert eq (counter get c) 0
        bs  = ask batch int c [1, 2, 3, 4]
        assert eq (ag sum ints bs) 20

        c2  = new counter agent 100
        rel = relay int c c2 5
        assert str eq (format agent reply rel) "115" "relay int c (10+5=15) -> c2 (100+15=115)"

        st1 = stop agent c
        assert str eq (format agent status st1) "stopped" "stop agent returns AgentStopped"
        assert eq (is agent alive c) False
        assert str eq (format agent status (agent status c)) "stopped" "agent status after stop"
        assert str eq (format agent reply (ask int c 1)) "agent stopped" "ask int after stop"
        assert str eq (format agent reply (try poll int c)) "agent stopped" "try poll int after stop"
        assert str eq (format agent reply (ask int with timeout c 1 50)) "agent stopped" "ask int with timeout after stop"
        assert str eq (format agent reply (relay int c c2 1)) "agent stopped" "relay int after stop"
        c0  = new counter agent 0
        assert int equals (counter get c0) 0 "counter 0 initial"
        assert int equals (counter add c0 3) 3 "counter 0 add 3"
        stop agent c0
        stop agent c2
        assert eq (is agent alive c2) False

39.3.5. Ejemplo 5: Agente de bitácora append-only (new log agent) y almacén clave-valor en memoria por paso de mensajes (new kv agent)

En este ejemplo práctico exploramos de forma exhaustiva cómo se comporta pool en este bloque funcional, incluyendo entradas estándar, combinaciones de parámetros y valores frontera:

fn :: test log and kv agents => ()
      test log and kv agents =
        lg  = new log agent "\n"
        assert ok (is agent alive lg)
        assert str eq (format agent reply (try poll str lg)) "agent outbox empty" "log empty outbox"
        assert str eq (log read lg) "" "initial log empty"
        assert str eq (log append lg "alpha") "alpha" "log append alpha"
        assert str eq (log append lg "beta") "alpha\nbeta" "log append beta"
        assert str eq (format agent reply (ask str with timeout lg "gamma" 200)) "alpha\nbeta\ngamma" "ask str with timeout gamma"
        assert str eq (log read lg) "alpha\nbeta\ngamma" "log read"
        assert str eq (log clear lg) "" "log clear"
        assert str eq (log read lg) "" "log read after clear"
        b   = ask batch str lg ["one", "two"]
        assert str eq (ag join strs b) "one|one\ntwo" "ask batch str"
        lg2 = new log agent ":"
        assert str eq (format agent reply (relay str lg lg2 "three")) "one\ntwo\nthree" "relay str"
        stop agent lg
        stop agent lg2
        assert str eq (format agent reply (ask str lg "x")) "agent stopped" "ask str after stop"

        kv  = new kv agent
        assert ok (is agent alive kv)
        assert int equals (kv count kv) 0 "kv count 0"
        assert false (kv has kv "host") "kv has host false"
        assert str eq (kv put kv "host" "127.0.0.1") "127.0.0.1" "kv put host"
        assert str eq (kv put kv "port" "8080") "8080" "kv put port"
        assert int equals (kv count kv) 2 "kv count 2"
        assert ok (kv has kv "host")
        assert ok (kv has kv "port")
        assert str eq (kv get kv "host") "127.0.0.1" "kv get host"
        assert str eq (kv get kv "port") "8080" "kv get port"
        assert str eq (kv keys kv) "host,port" "kv keys"
        assert str eq (kv put kv "port" "9090") "9090" "kv overwrite port"
        assert str eq (kv get kv "port") "9090" "kv get updated port"
        assert int equals (kv count kv) 2 "kv count after overwrite"
        assert str eq (kv delete kv "host") "127.0.0.1" "kv delete host"
        assert false (kv has kv "host") "kv has host after delete"
        assert int equals (kv count kv) 1 "kv count 1"
        assert str eq (kv clear kv) "" "kv clear"
        assert int equals (kv count kv) 0 "kv count after clear"
        stop agent kv
        assert false (is agent alive kv) "kv stopped"

39.3.6. Ejemplo 6: Enrutamiento determinista por clave (pool ask shard), difusión masiva (pool broadcast), drenaje no bloqueante (pool drain all) y apagado (pool shutdown)

En este ejemplo práctico exploramos de forma exhaustiva cómo se comporta pool en este bloque funcional, incluyendo entradas estándar, combinaciones de parámetros y valores frontera:

import agent
import pool

fn :: shard worker (state : Int) (delta : Int) => (Int, Int)
      shard worker state delta =
        next = state + delta
        (next, next)

fn :: ejemplo pool completo => (Int, Int, [Int])
      ejemplo pool completo =
        w0   = agent shard worker 100
        w1   = agent shard worker 200
        w2   = agent shard worker 300
        p    = Pool 3 [w0, w1, w2]
        sz   = pool size p
        -- Enrutamiento determinista por clave: abs(key_hash) % 3
        r_a  = pool ask shard p 0 5
        r_b  = pool ask shard p 1 10
        -- Difusión (broadcast) a todos los shards del pool
        pool broadcast p 1 -> ()
        pend = pool drain all p
        pool shutdown p -> ()
        (sz, r_a + r_b, pend)

39.4. Bajo el capó: cómo está construida pool en Ridgun puro (src/main/ridgun/lib/pool.rg)

Una de las mayores ventajas de ridgun-libs es que no hay cajas negras en C. Puedes abrir src/main/ridgun/lib/pool.rg y leer exactamente cómo está implementada cada operación usando pattern matching, árboles de guardas | / |. y recursión de cola (TCO). Veamos las piezas clave de su implementación real:

39.4.1. Implementación interna: Core Implementation Part 1

Fíjate en cómo pool.rg estructura las firmas fn :: ... => ... primero y resuelve cada caso mediante ecuaciones alineadas y llamadas de cola:

module :: pool

type AgentPool = Pool Int [Agent]

fn :: pool size AgentPool   => Int
      pool size (Pool sz _) =  sz

fn :: pool agents AgentPool    => [Agent]
      pool agents (Pool _ ags) =  ags

fn :: pool get agent [Agent]  Int => Agent
      pool get agent (ag:_)   0   =  ag
      pool get agent (_:rest) idx =  pool get agent rest (idx - 1)

fn :: pool abs Int => Int
      pool abs x
        | x < 0     = 0 - x
        | otherwise = x

fn :: pool ask shard AgentPool Int A => B
      pool ask shard (Pool size agents) key_hash msg =
        (pool abs key_hash) % size -> shard_idx
        pool get agent agents shard_idx -> target_ag
        ask target_ag msg

fn :: pool broadcast AgentPool       A   => ()
      pool broadcast (Pool _ agents) msg =  pool broadcast list agents msg

39.4.2. Implementación interna: Core Implementation Part 2

Fíjate en cómo pool.rg estructura las firmas fn :: ... => ... primero y resuelve cada caso mediante ecuaciones alineadas y llamadas de cola:

fn :: pool broadcast list [Agent]   A   => ()
      pool broadcast list []        _   =  ()
      pool broadcast list (ag:rest) msg =
        send msg to ag
        pool broadcast list rest msg

fn :: pool drain ready Agent => [B]
      pool drain ready ag    =  pool drain ready loop ag []

fn :: pool drain ready loop Agent [B] => [B]
      pool drain ready loop ag acc =
        pool drain step ag (try receive from ag) acc

fn :: pool drain step Agent Option [B] => [B]
      pool drain step _ None acc = acc
      pool drain step ag (Some r) acc = pool drain ready loop ag (acc :: [r])

fn :: pool drain all AgentPool       => [B]
      pool drain all (Pool _ agents) =  pool drain list agents []

fn :: pool drain list [Agent]   [B] => [B]
      pool drain list []        acc =  acc
      pool drain list (ag:rest) acc =
        ag pool drain ready -> ready
        pool drain list rest (acc :: ready)

fn :: pool stop list [Agent]   => ()
      pool stop list []        =  ()
      pool stop list (ag:rest) =
        stop ag
        pool stop list rest

fn :: pool shutdown AgentPool       => ()
      pool shutdown (Pool _ agents) =  pool stop list agents

39.5. Garantías algorítmicas, memoria RMM1/RMM2 y las 5 Reglas de Oro de pool

  • Garantía algorítmica y de seguridad: Capacidad de cola estrictamente acotada (queued <= max_queue) para proteger la memoria bajo sobrecarga y reparto determinista entre trabajadores.
  • Complejidad de pila $O(1)$ con TCO: Todos los bucles recursivos en src/main/ridgun/lib/pool.rg están en posición de cola, por lo que ridgunc los compila a saltos nativos sin crecimiento de pila.
  • Memoria determinista (RMM1/RMM2): Cero pausas de recolector de basura; todas las estructuras intermedias se liberan de forma determinista al salir del ámbito.
  • Verificación continua: Probada exhaustivamente con stable check y la suite automatizada src/test/ridgun/test pool.rg.

Las 5 Reglas de Oro al programar con pool

  1. Declara import pool al inicio del archivo y aprovecha la equivalencia entre espacios y guiones bajos para que tu código se lea como prosa.
  2. Alinea tus tuberías de datos con =: encadena las llamadas a pool en pasos con nombre propio y deja que stable format alinee las columnas.
  3. Aprovecha las funciones totales de pool en tus guardas |: comprueba los valores de retorno o desenvuelve los ADTs con pattern matching en lugar de asumir que la entrada externa siempre es válida.
  4. Evita copiar grandes estructuras si solo necesitas consultar un campo: recuerda que en RMM2 los argumentos de entrada no incrementan ni decrementan contadores (+0 / -0).
  5. Combina pool con test (import test): escribe siempre aserciones con assert equals o assert true para fijar los invariantes de tu dominio.

En el Capítulo 40 pasaremos a explorar en profundidad la lib net_agent (La lib net_agent — Supervisor de Conexiones de Red y Puente de Eventos No Bloqueante).