Travail distribué

Source: 06_distributed_work

Un serveur HTTP qui achemine chaque corps de requête POST /process vers un nœud worker distant, qui applique une transformation de texte ligne par ligne et renvoie le résultat en flux. Le worker est provisionné à la demande avec DistantEngine ; aucun second processus n’a besoin d’être démarré manuellement.

Exécution


melodium run 06_distributed_work/Compo.toml \
  --api_token "my-api-token"

Note

api_token authentifie ici auprès d’une API Mélodium Services, comme Cadence.CI .


$ curl -X POST http://127.0.0.1:8080/process -d "hello world"
[WORKER] hello world

Fonctionnement

Cet exemple introduit les deux primitives distribuées clés :

DistantEngine : contacte l’API Mélodium Services et provisionne un runner cloud
DistributionEngine : se connecte au runner et route le travail vers un traitement nommé


treatment main(
    const api_token: string,
    const port:      u16 = 8080
)
  model runner: DistantEngine(api_url=|wrap<string>("https://api.melodium.tech/0.1"), api_token=|wrap<string>(api_token))
  model distributor: DistributionEngine(
    treatment = "distributed_work/worker::process",
    version   = "0.1.0"
  )
  model server: HttpServer(
    host = |from_ipv4(|localhost_ipv4()),
    port = port
  )
{
    startup()
    logStart: logInfoMessage(label="cloud", message="provisioning worker runner…")
    startup.trigger -> logStart.trigger
 
    provisionRunner: distant[distant_engine=runner](
        max_duration = 600,
        memory       = 256, // Mo
        cpu          = 500, // millicœurs
        storage      = 256, // Mo
        edition      = _,
        arch         = _,
        volumes      = [],
        containers   = [],
        service_containers = [],
        tags         = []
    )
    startup.trigger -> provisionRunner.trigger,access -> distribStart.access

Séquence de démarrage

Le serveur HTTP ne démarre que lorsque le moteur de distribution signale ready, ce qui signifie que le worker distant est connecté et que le traitement est accessible :


    distribStart[distributor=distributor](params=|dmap([]))
 
    logReady:     logInfoMessage(label="distrib", message="worker runner connected")
    logFailed:    logErrorMessage(label="distrib", message="distribution engine failed")
    distribStart.ready  -> logReady.trigger
    distribStart.failed -> logFailed.trigger
 
    start[http_server=server]()
    logServerReady: logInfoMessage(label="server", message="HTTP server ready")
    distribStart.ready -> start.trigger
    distribStart.ready -> logServerReady.trigger
 
    connection[http_server=server](method=|post(), route="/process")
    status:      emit<HttpStatus>(value=|ok())
    headers:     emit<StringMap>(value=|map([]))
    bodyTrigger: trigger<byte>()
 
    connection.data -> bodyTrigger.stream,start --> status.trigger,emit  -> connection.status
                       bodyTrigger.start --------> headers.trigger,emit -> connection.headers
 
    dispatchToWorker[distributor=distributor]()
    connection.data -> dispatchToWorker.data,data -> connection.data
}

main treatment diagram Voir dans Compositeur Studio

Acheminement du travail

dispatchToWorker alloue un distribution_id par requête via distribute, puis utilise sendStream et recvStream nommés pour échanger des flux d’octets avec le traitement distant :


treatment dispatchToWorker[distributor: DistributionEngine]()
  input  data: Stream<byte>
  output data: Stream<byte>
{
    bodyTrigger: trigger<byte>()
    distribute[distributor=distributor]()
 
    Self.data -> bodyTrigger.stream,start -> distribute.trigger
 
    sendStream<byte>[distributor=distributor](name="data")
    recvStream<byte>[distributor=distributor](name="data")
 
    distribute.distribution_id -> sendStream.distribution_id
    distribute.distribution_id -> recvStream.distribution_id
 
    Self.data       -> sendStream.data
    recvStream.data -> Self.data
}

Les noms des flux ("data") doivent correspondre aux noms des ports d’entrée/sortie du traitement distant.

Le worker (`worker.mel`)

Le fichier worker définit deux traitements : runWorker (le point d’entrée, journalise un message de prêt) et process (la transformation réelle) :


treatment runWorker() {
    startup()
    log: logInfoMessage(label="worker", message="worker node ready")
    startup.trigger -> log.trigger
}
 
treatment process()
  input  data: Stream<byte>
  output data: Stream<byte>
{
    decode()
    split(delimiter="\n", inclusive=false)
    flatten<string>()
    wrapEntry: entry(key="line")
    fmt:       format(format="[WORKER] {line}")
    encode()
 
    Self.data -> decode.data,text -> split.text,splitted -> flatten.vector,value -> wrapEntry.value,map -> fmt.entries,formatted -> encode.text,data -> Self.data
}

process divise le flux d’octets entrant en lignes via split et flatten, préfixe chacune avec [WORKER] via entry et format, et réencode le résultat.

Dépendances


[dependencies]
std      = "0.10.1"  # flux de base, journalisation, structures de données
http     = "0.10.1"  # serveur et client HTTP
net      = "0.10.1"  # utilitaires d'adresses IP
encoding = "0.10.1"  # encodage / décodage UTF-8
distrib  = "0.10.1"  # distribution de flux entre runners
work     = "0.10.1"  # provisionnement de runners cloud