12.6 Distributed computation

oxl

let byRegion = cluster.shards(db.Order)
  .map(o => (o.region, o.total))          // executed on the shard owners
  .reduceBy(.0, (a, b) => a + b)          // partial aggregation, then merge
  within REPORT_DEADLINE

cluster.leader("nightly-report") { every(REPORT_PERIOD) { buildReport() } }
cluster.lock("user:{uid}", ttl: LOCK_TTL) { migrateUser(uid) }

cluster.shards(src) is for sharded tables and actors; map runs close to the data, the function MUST be transferable (§7.3).

Implementation-defined. Computation near the data (D-8055; DIV-015; D-4804, D-4899). An optional feature: the chain cluster.shards(src).map(…).reduceBy(…) within d. This implementation does not build it (a transferable closure and scatter-gather are needed); cluster.shards() is available — a list of shard groups ShardInfo. A query over all shards is executed by the ORM (§12.8). A possible extension — after 1.0.

cluster.leader(name) { … } registers a leader task and returns control immediately. On nodes whose roles contain runs leader(name), the nodes elect one leader and the block runs on it in the background scope; if the leader fails, the block is started on another node. On the other nodes the call does nothing.
cluster.lock(key, ttl) { … } is a distributed lock with a lease. The type of the call is T | LockTimeout | NetError, where T is the block type: LockTimeout — the lock could not be acquired within the waiting time; NetError (Unavailable) — the coordinator was unreachable in the last attempt of the wait. The same for cluster.lockFenced. Ignoring the result is the error unhandled-error (D-8052; D-7520).
cluster.ids.next() -> u64 — a per-node unique monotonic identifier; cluster.now() — a hybrid logical clock (§17).
cluster.self.hasRole(r) -> bool — whether the current node has the role r.
@near(expr) on a function executes it on the node where the actor or shard expr resides.

12.6 Distributed computation

See also