Traitement de texte distribué

Source: 13_distributed_text_processing

Un serveur HTTP qui reçoit du texte brut en entrée et le retourne en majuscules. La transformation s’exécute sur un runner Mélodium cloud provisionné à la demande ; le package javascript n’a besoin d’être disponible que sur le runner, pas sur la machine locale.

Exécution


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

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

Fonctionnement

Trois modèles sont instanciés dans main : le DistantEngine qui provisionne le runner cloud, le DistributionEngine qui route le travail vers celui-ci, et le HttpServer local :


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)

Le DistributionEngine référence processText par son chemin complet. processText (ainsi que le modèle Uppercaser qu’il utilise) est défini dans le même fichier main.mel, mais il ne s’exécute jamais que sur le runner distant.

Provisionnement et verrouillage du démarrage

distant demande le runner ; sa sortie access alimente start, et ce n’est qu’une fois le moteur de distribution passé à ready que le serveur HTTP commence réellement à accepter des connexions :


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

Aucun paramètre n’est nécessaire côté distant ici, donc start est appelé avec un params=|map([]) vide.

main treatment diagram Voir dans Compositeur Studio

Distribution de chaque requête

Chaque corps de requête POST /process est transmis en flux vers le traitement distant, et le résultat renvoyé en flux via dispatchProcess, qui encapsule distribute, sendStream et 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
}

Le flux envoyé et le flux reçu utilisent tous deux le nom "data", correspondant aux ports input data / output data de processText.

Le traitement distant `processText`

processText instancie son propre modèle Uppercaser, un JavaScriptEngine avec une fonction toUpper embarquée :


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
}

Il convertit le flux d’octets brut en chaîne via decode, l’enveloppe en JSON avec fromString<string>(), le fait passer par process, puis désenveloppe le résultat JSON en chaîne simple avant de le réencoder en octets. L’enveloppement/désenveloppement JSON est nécessaire car JavaScriptEngine.process échange des valeurs Json, pas des chaînes brutes.

processText treatment diagram Voir dans Compositeur Studio

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
javascript = "0.10.1"  # moteur JavaScript embarqué
json       = "0.10.1"  # parsing et sérialisation JSON
work       = "0.10.1"  # provisionnement de runners cloud
distrib    = "0.10.1"  # distribution de flux entre runners