39 Boken om Ridgun
Lib:en pool — Begränsad Worker-pool, Round-Robin och Work-Stealing
39.1. Varför finns lib:en pool? (Det verkliga problemet och Ridguns filosofi)
”Varför behöver jag en
pool-lib skriven från grunden i ren Ridgun i stället för att bara länka in ett gammalt C-bibliotek?”
Det är den första fråga varje systemingenjör ställer sig när man möter pool. Och svaret sammanfattar hela filosofin bakom ridgun-libs:
Låt oss vara ärliga: att starta en ny tråd eller aktör för varje inkommande anrop utan övre gräns garanterar att en trafiktopp tömmer serverns minne på några sekunder.
Lib:en pool tillhandahåller en begränsad uppgiftsfördelare med explicit mottryck (PoolAccept, PoolRejectFull), balanserad uppdelning i $K$ filer (pool partition), ombalansering med work-stealing (pool steal rebalance) och parallell exekvering över inbyggda aktörer.
I ett traditionellt ekosystem innebär uppgifter inom domänen AGENT ofta att man drar in externa C-beroenden, brottas med råa pekare, ogenomskinliga typkonverteringar, okontrollerade undantag eller oförutsägbara skräpsamlarpauser. I Ridgun löser lib:en pool (src/main/ridgun/lib/pool.rg) problemet genom fyra kompromisslösa principer:
- Deklarera
import pool(ellerimport pool as alias) överst i filen: varje lib definierar sinmodule :: pool-namnrymd och i--strict-läge anropas flerordsfunktioner alltid med naturliga mellanslag. - Kedja dina datapipelines från vänster till höger med
->: skrivindata steg ett -> steg två -> resultat, använd(+ _ _ +)-sekvensmasker för att packa upp tupler och utelämna_ =vid sekventiella satser (eller kasta med-> ()). - Svansrekursion med TCO (Tail-Call Optimization): alla interna genomlöpningar i
poolkompileras till inbyggda hårdvaruloopar med $O(1)$ stackutrymme. - Deterministiskt minne (
RMM1/RMM2): varje mellanliggande struktur frigörs i exakt det ögonblick dess sista läsning avslutas, helt utan Garbage Collector.
39.2. Inbyggda artefakter, import och komplett API-katalog för pool
I ridgun-libs är pool en förstklassig lib placerad i src/main/ridgun/lib/pool.rg (domän AGENT). När du bygger med stable bundle paketeras den i tre inbyggda format:
| Artefakt | Sökväg | Syfte |
|---|---|---|
| Ridgun-bundle | target/bundles/pool.rglib | Portabel distribution och direkt import med import pool |
| Kompilerat arkiv | target/pool.rgc | Typkontrollerad mellanartefakt verifierad av ridgunc |
| Delat ELF64-objekt | target/libpool.so | Inbyggt dynamiskt bibliotek helt utan C- eller libc-beroenden |
Varje lib deklarerar sin egen namnrymd med
module :: pool: skriv baraimport pool(ellerimport pool as alias) överst och anropa dess funktioner med naturliga mellanslag (min funktion x) och->-pipelines (x min funktion -> resultat).
39.2.1. Översikt över funktionella familjer i pool
| Signatur / Funktion / Typ | Vad den gör och när den används |
|---|---|
pool new : Int -> Int -> PoolState | Skapar en begränsad pool med workers filer och max_queue i kökapacitet |
pool submit : PoolState -> PoolTask -> (PoolState, PoolDecision) | Köar en uppgift med round-robin eller returnerar PoolRejectFull vid mättnad |
pool partition / pool steal rebalance | Delar upp batcher över $K$ arbetare och stjäl hälften av en överbelastad kö till en ledig arbetare |
pool map sum / pool parallel sum2 | Fördelar arbetslaster parallellt över inbyggda agent-arbetare |
39.2.2. Komplett katalog över typer och signaturer i src/main/ridgun/lib/pool.rg
Nedan följer hela förteckningen över algebraiska datatyper (type), mixfix-operatorer (op ::) och funktionssignaturer (fn ::) som definieras i src/main/ridgun/lib/pool.rg (1 typer, 0 operatorer och 14 funktioner):
Algebraiska datatyper och poster (type) i pool:
type AgentPool = Pool Int [Agent]
| # | Funktion / Signatur (fn ::) | Returtyp |
|---|---|---|
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. Djupdykning i pool: Steg-för-steg-exempel för varje funktion
Det räcker inte med en signaturtabell: låt oss utforska varje funktionalitet i lib:en pool genom progressiva, körbara Ridgun-exempel. Vi börjar med två tabellbaserade vardagsrecept och går därefter igenom alla funktionella scenarier, transformationer och gränsfall som verifieras i src/test/ridgun/test pool.rg.
39.3.1. Exempel 1: Huvudflöde i tabellform med pool
Lägg märke till hur =-tecknen och transformationsstegen med pool linjeras som en ren tabell utan onödiga muterbara variabler eller dolda sidoeffekter:
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. Exempel 2: Säker komposition och hantering av gränsfall
Varje funktion i lib:en pool är utformad för att kombineras sömlöst med booleska vakter | och med övriga 32 libs i 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. Exempel 3: Aktörskombinatorer (agent ask, try ask, send batch, map batch, drain ready), shardade pooler och anslutningsreaktor
I detta praktiska exempel undersöker vi grundligt hur pool beter sig inom detta funktionsområde, inklusive standardindata, parameterkombinationer och gränsvärden:
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. Exempel 4: Agentlivscykel (is agent alive, agent status, stop agent) och isolerad tillståndsfull räknaragent (new counter agent)
I detta praktiska exempel undersöker vi grundligt hur pool beter sig inom detta funktionsområde, inklusive standardindata, parameterkombinationer och gränsvärden:
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. Exempel 5: Append-only-loggagent (new log agent) och meddelandebaserad nyckel-värde-lagring i minnet (new kv agent)
I detta praktiska exempel undersöker vi grundligt hur pool beter sig inom detta funktionsområde, inklusive standardindata, parameterkombinationer och gränsvärden:
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. Exempel 6: Deterministisk nyckelshardning (pool ask shard), utsändning (pool broadcast), icke-blockerande tömning (pool drain all) och avstängning (pool shutdown)
I detta praktiska exempel undersöker vi grundligt hur pool beter sig inom detta funktionsområde, inklusive standardindata, parameterkombinationer och gränsvärden:
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. Under huven: hur pool är byggd i ren Ridgun (src/main/ridgun/lib/pool.rg)
En av de största fördelarna med ridgun-libs är att det inte finns några svarta lådor i C. Du kan öppna src/main/ridgun/lib/pool.rg och läsa exakt hur varje operation är implementerad med mönstermatchning, | / |.-vaktträd och svansrekursion (TCO). Låt oss titta på nyckeldelarna i den faktiska implementationen:
39.4.1. Intern implementation: Core Implementation Part 1
Se hur pool.rg deklarerar explicita fn :: ... => ...-kontrakt först och löser varje fall genom linjerade ekvationer och svansanrop:
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. Intern implementation: Core Implementation Part 2
Se hur pool.rg deklarerar explicita fn :: ... => ...-kontrakt först och löser varje fall genom linjerade ekvationer och svansanrop:
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. Algoritmiska garantier, RMM1/RMM2-minne och de 5 Gyllene Reglerna för pool
- Algoritmisk garanti och säkerhet: Strikt begränsad kökapacitet (
queued <= max_queue) för att skydda minnet vid belastningstoppar samt deterministisk fördelning mellan arbetare. - $O(1)$ stackkomplexitet med TCO: Alla rekursiva loopar i
src/main/ridgun/lib/pool.rgligger i svansposition, såridgunckompilerar dem till inbyggda hopp utan stacktillväxt. - Deterministiskt minne (
RMM1/RMM2): Noll skräpsamlarpauser; alla temporära strukturer frigörs deterministiskt när räckvidden lämnas. - Kontinuerlig verifiering: Statiskt kontrollerad med
stable checkoch verifierad av den automatiserade testsvitensrc/test/ridgun/test pool.rg.
De 5 Gyllene Reglerna när du programmerar med pool
- Deklarera
import poolöverst i filen och utnyttja att mellanslag och understreck är utbytbara så att koden läses som naturlig prosa. - Linjera dina datapipelines med
=: kedja sammanpool-anrop i namngivna mellansteg och låtstable formatlinjera kolumnerna. - Utnyttja
pool:s totala funktioner i dina|-vakter: kontrollera returvärden eller packa upp ADT:er med mönstermatchning i stället för att anta att extern indata alltid är giltig. - Undvik att kopiera stora strukturer när du bara läser ett fält: kom ihåg att i
RMM2medför inargument noll referensräkningskostnad (+0/-0). - Kombinera
poolmedtest(import test): lås alltid fast dina domäninvarianter medassert equalsellerassert true.
I Kapitel 40 går vi vidare och utforskar lib:en net_agent på djupet (Lib:en net_agent — Övervakare för Nätverksanslutningar och Icke-blockerande Händelsebrygga).