39 O Libro de Ridgun
A lib pool — Pool Acoutado de Traballadores, Round-Robin e Roubo de Traballo
39.1. Por que existe a lib pool? (O problema real e a filosofía Ridgun)
«Por que necesito unha lib
poolescrita dende cero en Ridgun puro en vez de enlazar unha librería de C de toda a vida?»
Esa é a primeira pregunta que se fai calquera enxeñeiro de sistemas ao chegar a pool. E a resposta resume toda a filosofía de ridgun-libs:
Sexamos sinceros: lanzar un fío ou actor novo por cada petición entrante sen límite superior (unbounded spawning) fai que un pico de tráfico esgote a memoria en segundos (Out Of Memory).
A lib pool proporciona un despachador de tarefas con contrapresión explícita (PoolAccept, PoolRejectFull), partición equilibrada de lotes en $K$ carrís (pool partition), reequilibrio por roubo de traballo (work-stealing) (pool steal rebalance) e execución paralela sobre actores nativos.
Nun ecosistema tradicional, resolver as tarefas do dominio AGENT implica arrastrar dependencias externas en C, loitar con punteiros crus, conversións opacas de tipos, excepcións non verificadas ou pausas impredicibles do colector de lixo. En Ridgun, a lib pool (src/main/ridgun/lib/pool.rg) aborda o problema con catro principios innegociables:
- Declara
import pool(ouimport pool as alias) ao inicio do ficheiro: cada lib define o seu espazo de nomesmodule :: poole en modo--strictas funcións multi-palabra invócanse sempre con espazos naturais. - Encadea as túas tubaxes de esquerda a dereita con
->: escribeentrada paso un -> paso dous -> resultado, usa máscaras(+ _ _ +)para desestruturar secuencias e omite_ =en sentenzas secuenciais (ou descarta con-> ()). - Recursión de cola con TCO (Tail-Call Optimization): todos os percorridos internos de
poolcompílanse a bucles nativos en espazo de pila $O(1)$. - Memoria determinista (
RMM1/RMM2): cada estrutura intermedia libérase no instante exacto en que remata a súa última lectura, sen Garbage Collector.
39.2. Artefactos nativos, importación e catálogo completo da API de pool
En ridgun-libs, pool é unha lib de primeira clase situada en src/main/ridgun/lib/pool.rg (dominio AGENT). Ao compilar con stable bundle, empaquétase en tres formatos nativos:
| Artefacto | Ruta | Propósito |
|---|---|---|
| Bundle Ridgun | target/bundles/pool.rglib | Distribución e importación directa con import pool |
| Ficheiro compilado | target/pool.rgc | Artefacto intermedio verificado por ridgunc |
| Obxecto compartido ELF64 | target/libpool.so | Biblioteca dinámica nativa sen dependencias de C nin libc |
Cada lib declara o seu propio espazo de nomes con
module :: pool: abonda con escribirimport pool(ouimport pool as alias) na cabeceira e chamar ás súas funcións con espazos naturais (mina funcion x) e tubaxes->(x mina funcion -> resultado).
39.2.1. Resumo de familias funcionais de pool
| Sinatura / Función / Tipo | Que fai e cando usala |
|---|---|
pool new : Int -> Int -> PoolState | Crea un pool acoutado con workers carrís e capacidade máxima max_queue |
pool submit : PoolState -> PoolTask -> (PoolState, PoolDecision) | Enfila unha tarefa en round-robin ou devolve PoolRejectFull se está saturado |
pool partition / pool steal rebalance | Reparte lotes entre $K$ traballadores e rouba a metade dunha fila sobrecargada para un traballador libre |
pool map sum / pool parallel sum2 | Despacha cargas de traballo en paralelo sobre traballadores agent nativos |
39.2.2. Catálogo exhaustivo de tipos e sinaturas en src/main/ridgun/lib/pool.rg
A continuación tes o inventario completo de tipos alxébricos (type), operadores (op ::) e sinaturas de funcións (fn ::) definidos en src/main/ridgun/lib/pool.rg (1 tipos, 0 operadores e 14 funcións):
Tipos alxébricos e rexistros (type) de pool:
type AgentPool = Pool Int [Agent]
| # | Función / Sinatura (fn ::) | Tipo de Retorno |
|---|---|---|
1 | fn :: pool size AgentPool | Int |
2 | fn :: pool agents AgentPool | [Agent] |
3 | fn :: pool get agent [Agent] Int | Agent |
4 | fn :: pool abs Int | Int |
5 | fn :: pool ask shard AgentPool Int A | B |
6 | fn :: pool broadcast AgentPool A | () |
7 | fn :: pool broadcast list [Agent] A | () |
8 | fn :: pool drain ready Agent | [B] |
9 | fn :: pool drain ready loop Agent [B] | [B] |
10 | fn :: pool drain step Agent Option [B] | [B] |
11 | fn :: pool drain all AgentPool | [B] |
12 | fn :: pool drain list [Agent] [B] | [B] |
13 | fn :: pool stop list [Agent] | () |
14 | fn :: pool shutdown AgentPool | () |
39.3. Explorando pool a fondo: Exemplos paso a paso de cada funcionalidade
Non abonda con ver a táboa de sinaturas: imos explorar cada unha das funcionalidades da lib pool mediante exemplos progresivos e executables en Ridgun. Comezaremos polas dúas receitas tabulares de uso cotián e despois percorreremos en detalle todos os escenarios funcionais, transformacións e casos límite verificados en src/test/ridgun/test pool.rg.
39.3.1. Exemplo 1: Fluxo principal en estilo tabular con pool
Observa como en Ridgun os signos = e os pasos de transformación con pool quedan aliñados coma unha táboa limpa, sen variables mutables innecesarias nin efectos laterais 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. Exemplo 2: Composición segura e manexo de casos de bordo
Toda función da lib pool está deseñada para compoñerse con gardas booleanas | e co resto das 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. Exemplo 3: Combinadores de actores (agent ask, try ask, send batch, map batch, drain ready), pools particionados e reactor de conexións
Neste exemplo práctico exploramos de forma exhaustiva como se comporta pool neste bloque funcional, incluíndo entradas estándar, combinacións de parámetros e valores fronteira:
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. Exemplo 4: Ciclo de vida de axentes (is agent alive, agent status, stop agent) e axente contador con estado illado (new counter agent)
Neste exemplo práctico exploramos de forma exhaustiva como se comporta pool neste bloque funcional, incluíndo entradas estándar, combinacións de parámetros e valores fronteira:
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. Exemplo 5: Axente de bitácora append-only (new log agent) e almacén clave-valor en memoria por paso de mensaxes (new kv agent)
Neste exemplo práctico exploramos de forma exhaustiva como se comporta pool neste bloque funcional, incluíndo entradas estándar, combinacións de parámetros e valores fronteira:
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. Exemplo 6: Enrutamento determinista por clave (pool ask shard), difusión masiva (pool broadcast), drenaxe non bloqueante (pool drain all) e apagado (pool shutdown)
Neste exemplo práctico exploramos de forma exhaustiva como se comporta pool neste bloque funcional, incluíndo entradas estándar, combinacións de parámetros e valores fronteira:
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. Baixo o capó: como está construída pool en Ridgun puro (src/main/ridgun/lib/pool.rg)
Unha das maiores vantaxes de ridgun-libs é que non hai caixas negras en C. Podes abrir src/main/ridgun/lib/pool.rg e ler exactamente como está implementada cada operación usando pattern matching, árbores de gardas | / |. e recursión de cola (TCO). Vexamos as pezas clave da súa implementación real:
39.4.1. Implementación interna: Core Implementation Part 1
Fíxate en como pool.rg estrutura as sinaturas fn :: ... => ... primeiro e resolve cada caso mediante ecuacións aliñadas e chamadas 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íxate en como pool.rg estrutura as sinaturas fn :: ... => ... primeiro e resolve cada caso mediante ecuacións aliñadas e chamadas 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 e as 5 Regras de Ouro de pool
- Garantía algorítmica e de seguridade: Capacidade de fila estritamente acoutada (
queued <= max_queue) para protexer a memoria baixo sobrecarga e reparto determinista entre traballadores. - Complexidade de pila $O(1)$ con TCO: Todos os bucles recursivos en
src/main/ridgun/lib/pool.rgestán en posición de cola, polo queridguncos compila a saltos nativos sen crecemento de pila. - Memoria determinista (
RMM1/RMM2): Cero pausas de colector de lixo; todas as estruturas intermedias libéranse de forma determinista ao saír do ámbito. - Verificación continua: Probada exhaustivamente con
stable checke a suite automatizadasrc/test/ridgun/test pool.rg.
As 5 Regras de Ouro ao programar con pool
- Declara
import poolao inicio do ficheiro e aproveita a equivalencia entre espazos e guións baixos para que o teu código se lea coma prosa. - Aliña as túas tubaxes de datos con
=: encadea as chamadas apoolen pasos con nome propio e deixa questable formataliñe as columnas. - Aproveita as funcións totais de
poolnas túas gardas|: comproba os valores de retorno ou desenvolve os ADTs con pattern matching en vez de asumir que a entrada externa sempre é válida. - Evita copiar grandes estruturas se só necesitas consultar un campo: lembra que en
RMM2os argumentos de entrada non incrementan nin decrementan contadores (+0/-0). - Combina
poolcontest(import test): escribe sempre asercións conassert equalsouassert truepara fixar os invariantes do teu dominio.
No Capítulo 40 pasaremos a explorar en profundidade a lib net_agent (A lib net_agent — Supervisor de Conexións de Rede e Ponte de Eventos Non Bloqueante).