Distributed Work

Source: 06_distributed_work

An HTTP front-end that dispatches each POST /process request body to a remote worker node, which applies a line-level text transformation and streams the result back. The worker is provisioned on demand using DistantEngine; no second process needs to be started manually.

Running


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

Note

api_token here authenticates against a Mélodium Services API, such as Cadence.CI .


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

How it works

This example introduces the two key distributed primitives:

DistantEngine: contacts the Mélodium Services API and provisions a cloud runner
DistributionEngine: connects to the runner and routes work to a named treatment


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, // MB
        cpu          = 500, // millicores
        storage      = 256, // MB
        edition      = _,
        arch         = _,
        volumes      = [],
        containers   = [],
        service_containers = [],
        tags         = []
    )
    startup.trigger -> provisionRunner.trigger,access -> distribStart.access

Startup sequence

The HTTP server only starts once the distribution engine signals ready, meaning the remote worker is connected and the treatment is reachable:


    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 See in Compositeur Studio

Dispatching work

dispatchToWorker allocates a distribution_id per request using distribute, then uses named sendStream and recvStream to exchange byte streams with the remote treatment:


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
}

The stream names ("data") must match the input/output port names of the remote treatment.

The worker (`worker.mel`)

The worker file defines two treatments: runWorker (the entrypoint, logs a ready message) and process (the actual transformation):


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 splits the incoming byte stream into lines using split and flatten, prepends [WORKER] to each using entry and format, and re-encodes the result.

Dependencies


[dependencies]
std      = "0.10.1"  # core flows, logging, data structures
http     = "0.10.1"  # HTTP server and client
net      = "0.10.1"  # IP address helpers
encoding = "0.10.1"  # UTF-8 encode / decode
distrib  = "0.10.1"  # stream distribution across runners
work     = "0.10.1"  # cloud runner provisioning