Distributed Text Processing

Source: 13_distributed_text_processing

An HTTP server that receives plain-text input and returns it uppercased. The transformation runs on a Mélodium cloud runner provisioned on demand; the javascript package only needs to be available on the runner, not the front-end machine.

Running


melodium run 13_distributed_text_processing/Compo.toml \
  --api_token "my-api-token" \
  --port 8080

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 \
       -H "Content-Type: text/plain" \
       -d "hello world from melodium"
HELLO WORLD FROM MELODIUM

How it works

Three models are instantiated in main: the DistantEngine that provisions the cloud runner, the DistributionEngine that routes work to it, and the local HttpServer:


model runner: DistantEngine(api_url=|wrap<string>("https://api.melodium.tech/0.1"), api_token=|wrap<string>(api_token))
model distributor: DistributionEngine(
  treatment = "distributed_text_processing/main::processText",
  version   = "0.1.0"
)
model server: HttpServer(host=|from_ipv4(|localhost_ipv4()), port=port)

DistributionEngine references processText by its fully-qualified path. processText (and the Uppercaser model it uses) are defined in the same main.mel file, but they only ever execute on the remote runner.

Provisioning and startup gating

distant requests the runner; its access output feeds start, and only once the distribution engine reports ready does the HTTP server actually start accepting connections:


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
 
distribStart: start[distributor=distributor](params=|map([]))
 
startHttp[http_server=server]()
distribStart.ready -> startHttp.trigger

No parameters are needed on the remote side here, so start is called with an empty params=|map([]).

main treatment diagram See in Compositeur Studio

Dispatching each request

Each POST /process body is streamed to the remote treatment and the result streamed back through dispatchProcess, which wraps distribute, sendStream, and recvStream:


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

Both the sent and received streams use the name "data", matching processText’s own input data / output data port names.

The remote `processText` treatment

processText instantiates its own Uppercaser model, a JavaScriptEngine with an embedded toUpper function:


model Uppercaser() : JavaScriptEngine {
    code = "function toUpper(text) { return text.toString().toUpperCase(); }"
}
 
treatment processText()
  model uppercaser: Uppercaser()
  input  data: Stream<byte>
  output data: Stream<byte>
{
    decode()
    wrapStr:      fromString<string>()
    jsUpper:      process[engine=uppercaser](code="toUpper(value)")
    unwrapResult: unwrapOr<Json>(default=|null())
    resultStr:    tryToString<Json>()
    unwrapStr:    unwrapOr<string>(default="")
    encode()
 
    Self.data -> decode.data,text -> wrapStr.value,json -> jsUpper.value,result -> unwrapResult.option,value -> resultStr.value,into -> unwrapStr.option,value -> encode.text,data -> Self.data
}

It converts the raw byte stream to a string via decode, wraps it as JSON using fromString<string>(), runs it through process, then unwraps the JSON result back to a plain string before re-encoding it to bytes. The JSON wrapping/unwrapping is required because JavaScriptEngine.process exchanges Json values, not raw strings.

processText treatment diagram See in Compositeur Studio

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
javascript = "0.10.1"  # embedded JavaScript engine
json       = "0.10.1"  # JSON parsing and serialisation
work       = "0.10.1"  # cloud runner provisioning
distrib    = "0.10.1"  # stream distribution across runners