Skip to Content
Mélodium 0.11.0 is now available!
DocsExamples10. Distributed Computation

Distributed Computation

Source: tutorial/10_distributed_computation (directory) See in Playground

Every previous example ran in a single process; distribute is the one primitive that spreads a computation across several engines, the mechanism the showcase examples lean on for scaling further. A local client sends a stream of numbers to a second, separately running Mélodium engine; that engine doubles each number and sends it back.

Unlike every other tutorial step, this example is a single standalone .mel file, distributed_computation.mel, run directly with melodium run. There is no Compo.toml.

Note

Requires two terminals, each running its own Mélodium engine, sharing the same distribution group (a MELODIUM_GROUP_ID environment variable) and a matching pair of authentication keys.

Running

Both terminals must export the same MELODIUM_GROUP_ID. The two keys are swapped between them: one side’s send key is the other side’s recv key. The UUIDs below are example values for local pairing only, any matching pair works as long as both terminals agree.

Terminal 1, the listening engine:

cd tutorial/10_distributed_computation
export MELODIUM_GROUP_ID=10101010-1010-1010-1010-101010101010
export MELODIUM_DIST_RECV_KEY=11111111-1111-1111-1111-111111111111
export MELODIUM_DIST_SEND_KEY=22222222-2222-2222-2222-222222222222
melodium dist --localhost --port 6789

Terminal 2, this script, with the keys swapped:

cd tutorial/10_distributed_computation
export MELODIUM_GROUP_ID=10101010-1010-1010-1010-101010101010
export MELODIUM_SECRET_DIST_SEND_KEY=11111111-1111-1111-1111-111111111111
export MELODIUM_SECRET_DIST_RECV_KEY=22222222-2222-2222-2222-222222222222
melodium run distributed_computation.mel --port 6789
info: distrib: connected to remote engine
info: doubled: 6
info: doubled: 6
info: doubled: 6
info: doubled: 6
info: doubled: 6

--localhost uses an embedded certificate meant for local testing. The listener must be up first: main only sends data once it confirms the connection is ready, but that connection attempt can only succeed once the listener itself is accepting it.

Optional: add --api-report and an API token (MELODIUM_API_TOKEN) to see this run’s full trace on Cadence.CI .

How it works

A single DistributionEngine model identifies the remote treatment to run and its version, not a network resource:

model Doubler() : DistributionEngine {
    treatment = "distributed_computation::double"
    version   = "0.1.0"
}

This is different from models like SqlPool or HttpServer: the network target itself is supplied separately, through an Access value passed to start. double, the treatment executed on the remote engine, is defined in the same file:

treatment double()
  input  n: Stream<i64>
  output n: Stream<i64>
{
    doubled: add<i64>()
    Self.n -> doubled.a
    Self.n -> doubled.b
    doubled.sum -> Self.n
}

The overall data flow crosses the network in both directions:

Connecting to the remote engine

work/access::|new_access builds an Access value (IP, port, and the two authentication keys) entirely from parameters, no cloud service involved, just a second Mélodium process reachable over the network. start then opens the connection:

treatment main(
    const port:     u16    = 6789,
    const send_key: Secret<string> = "env:MELODIUM_SECRET_DIST_SEND_KEY",
    const recv_key: Secret<string> = "env:MELODIUM_SECRET_DIST_RECV_KEY",
    const amount:   u128   = 5,
    const value:    i64    = 3
)
  model distributor: Doubler()
{
    startup()
 
    accessBlock: emit<Access>(value=|new_access([|from_ipv4(|localhost_ipv4())], port, send_key, recv_key))
    startup.trigger -> accessBlock.trigger
 
    distribStart: start[distributor=distributor](params=|map([]))
    accessBlock.emit -> distribStart.access
 
    distribErr:    logError(label="distrib")
    distribFailed: logErrorMessage(label="distrib", message="could not connect to the remote engine")
    distribStart.error  -> distribErr.message
    distribStart.failed -> distribFailed.trigger
 
    logReady: logInfoMessage(label="distrib", message="connected to remote engine")
    distribStart.ready -> logReady.trigger
 
    run[distributor=distributor](amount=amount, value=value)
    distribStart.ready -> run.trigger
}

The two keys are secrets: send_key and recv_key are Secret<string> parameters defaulting to the locators env:MELODIUM_SECRET_DIST_SEND_KEY and env:MELODIUM_SECRET_DIST_RECV_KEY, and only distrib::start reveals them, to authenticate the connection. They never appear in logs, debug events, or reports. The listening engine reads its own keys from MELODIUM_DIST_RECV_KEY and MELODIUM_DIST_SEND_KEY (or from files given with --recv-key-file and --send-key-file), rather than from command arguments that other local users could read.

|new_access’s parameter order is (ip, port, remote_key, self_key): remote_key is the identity presented outward, the local send_key, and self_key is what is checked against what comes back, the local recv_key. Only once distribStart.ready fires does run actually build and send any data, so nothing races the connection setup. With the defaults (--amount 5 --value 3), the client sends five copies of 3 and logs doubled: 6 five times.

Sending and receiving across the network

dispatchDouble is the general shape for “run this like a local treatment, but remotely”: distribute allocates a distribution_id for one exchange, then sendStream and recvStream, tagged with matching names, carry the actual data in both directions:

treatment run[distributor: DistributionEngine](const amount: u128, const value: i64)
  input trigger: Block<void>
{
    length: emit<u128>(value=amount)
    numbers: generate<i64>(data=value)
    Self.trigger -> length.trigger,emit -> numbers.length,stream -> dispatch.n
 
    dispatch: dispatchDouble[distributor=distributor]()
 
    logResult: logInfos(label="doubled")
    asText: toString<i64>()
    dispatch.n -> asText.value,into -> logResult.messages
}
 
treatment dispatchDouble[distributor: DistributionEngine]()
  input  n: Stream<i64>
  output n: Stream<i64>
{
    startTrigger: trigger<i64>()
    dist: distribute[distributor=distributor]()
    Self.n -> startTrigger.stream,start -> dist.trigger
 
    send: sendStream<i64>[distributor=distributor](name="n")
    recv: recvStream<i64>[distributor=distributor](name="n")
    dist.distribution_id -> send.distribution_id
    dist.distribution_id -> recv.distribution_id
 
    Self.n    -> send.data
    recv.data -> Self.n
}

This is the three-step handshake for one remote call: allocate an ID, send the input, receive the output, all tagged by port name ("n" here) so multiple streams can cross the same connection unambiguously. From dispatchDouble’s own inputs and outputs, calling the remote double treatment looks exactly like connecting to a local one.

Dependencies

There is no Compo.toml for this example: as a standalone script, distributed_computation.mel declares its dependencies directly in its own header:

#!/usr/bin/env melodium
#! name    = distributed_computation
#! version = 0.1.0
#! require = std:0.11.* net:0.11.* work:0.11.* distrib:0.11.*
  • std: core flows, logging, data structures
  • net: IP address helpers
  • work: network access description for the remote engine
  • distrib: running treatments on a remote engine and wiring their streams
09. Process Pipeline11. Signed Webhooks