Parallel
Parallel execution is where Aqua fully shines.

Contract

  • Parallel arms have no access to each other's data. Sync points must be explicit (see Join behavior).
  • If any arm is executed successfully, the flow execution continues.
  • All the data defined in parallel arms is available in the subsequent code.

Implementation limitation

Parallel execution has some implementation limitations:
  • Parallel means independent execution on different peers
  • No parallelism when executing a script on a single peer
  • No concurrency in services: every service instance does only one job simultaneously.
  • Keep services small in terms of computation and memory (WebAssembly limitation)
These limitations might be overcome in future Aqua updates. But for now, plan your application design having this in mind.

Parallel operations

par

par syntax is derived from Ο€-calculus notation of parallelism: A | B
1
-- foo and bar will be executed in parallel, if possible
2
foo()
3
par bar()
4
​
5
-- It's useful to combine `par` with `on` block,
6
-- to delegate further execution to different peers.
7
​
8
-- In this case execution will continue on two peers, independently
9
on "peer 1":
10
x <- foo()
11
par on "peer 2":
12
y <- bar()
13
​
14
-- Once any of the previous functions return x or y,
15
-- execution continues. We don't know the order, so
16
-- if y is returned first, hello(x) will not execute
17
hello(x)
18
hello(y)
19
​
20
-- You can fix it with par
21
-- What's comes faster, will advance the execution flow
22
hello(x)
23
par hello(y)
Copied!
par works in an infix manner between the previously stated function and the next one.

co

co , short for coroutine, prefixes an operation to send it to the background. From Ο€-calculus perspective, it's the same as A | null, where null-process is the one that does nothing and completes immediately.
1
-- Let's send foo to background and continue
2
co foo()
3
​
4
-- Do something on another peer, not blocking the flow on this one
5
co on "some peer":
6
baz()
7
​
8
-- This foo does not wait for baz()
9
foo()
10
​
11
-- Assume that foo is executed on another machine
12
co try:
13
x <- foo()
14
-- bar will not wait for foo to be executed or even launched
15
bar()
16
-- bax will wait for foo to complete
17
-- if foo failed, x will never resolve
18
-- and bax will never execute
19
bax(x)
Copied!

Join behavior

Join means that data was created by different parallel execution flows and then used on a single peer to perform computations. It works the same way for any parallel blocks, be it par, co or something else (for par).
In Aqua, you can refer to previously defined variables. In case of sequential computations, they are available, if execution not failed:
1
-- Start execution somewhere
2
on peer1:
3
-- Go to peer1, execute foo, remember x
4
x <- foo()
5
​
6
-- x is available at this point
7
​
8
on peer2:
9
-- Go to peer2, execute bar, remember y
10
y <- bar()
11
​
12
-- Both x and y are available at this point
13
-- Use them in a function
14
baz(x, y)
Copied!
Let's make this script parallel: execute foo and bar on different peers in parallel, then use both to compute baz.
1
-- Start execution somewhere
2
on peer1:
3
-- Go to peer1, execute foo, remember x
4
x <- foo()
5
​
6
-- Notice par on the next line: it means, go to peer2 in parallel with peer1
7
​
8
par on peer2:
9
-- Go to peer2, execute bar, remember y
10
y <- bar()
11
​
12
-- Remember the contract: either x or y is available at this point
13
-- As it's enough to execute just one branch to advance further
14
baz(x, y)
Copied!
What will happen when execution comes to baz?
Actually, the script will be executed twice: the first time it will be sent from peer1, and the second time – from peer2. Or another way round: peer2 then peer1, we don't know who is faster.
When execution will get to baz for the first time, Aqua VM will realize that it lacks some data that is expected to be computed above in the parallel branch. And halt.
After the second branch executes, VM will be woken up again, reach the same piece of code and realize that now it has enough data to proceed.
This way you can express race (see Collection types and Conditional return for other uses of this pattern):
1
-- Initiate a stream to write into it several times
2
results: *string
3
​
4
on peer1:
5
results <- foo()
6
​
7
par on peer2:
8
results <- bar()
9
​
10
-- When any result is returned, take the first (the fastest) to proceed
11
baz(results!)
Copied!

Explicit join expression

Consider the case when you want to wait for a certain amount of results computed in parallel, and then return.
1
func collectRespondingPeers(peers: []string, n: i16) -> *string:
2
responded: *string
3
for p <- peers par:
4
responded <<- p
5
-- ...
Copied!
How to return no less than n+1 responsible peers?
Keep in mind that indices start from 0.
If the expected length of a stream equals n, and you wait for element stream[n], your code will hang forever, as it exceeds the length!
One way is to use a useless stream:
1
useless: *string
2
useless <<- responded[n]
3
<- responded
Copied!
Actually useless stream is useless, we create it just to push the nth element into it. However, it forces waiting for responded[n] to be available. When responded is returned, it will be at least of length n+1 or longer.
To eliminate the need for such workarounds, Aqua has the join expression that does nothing except consuming its arguments, hence waiting for them:
1
join responded[n]
2
<- responded
Copied!
You can use any number of arguments to join, separating them with a comma. join is executed on a particular node – so join respects the on scopes it's in.
1
func getTwo(peerA: string, peerB: string) -> string, string:
2
co on peerA:
3
a <- foo()
4
co on peerB:
5
b <- foo()
6
7
-- Without this join, a and b will still be returned,
8
-- But in the join case they are returned by-value,
9
-- While without join they are returned by-name
10
-- and it could happen that they're not actually available in time.
11
join a, b
12
<- a, b
Copied!

Timeout and race patterns

To limit the execution time of some part of an Aqua script, you can use a pattern that's often called "race". Execute a function in parallel with Peer.timeout, and take results from the first one to complete.
This way, you're racing your function against timeout. If timeout is the first one to complete, consider your function "timed out".
Peer.timeout is defined in aqua-lib.
For this pattern to work, it is important to keep an eye on where exactly the timeout is scheduled and executed. One caveat is that you cannot timeout the unreachable peer by calling a timeout on that peer.
Here's an example of how to put a timeout on peer traversal:
1
-- Peer.timeout comes from the standard library
2
import "@fluencelabs/aqua-lib/builtin"
3
​
4
func traverse_peers(peers: []string) -> []string:
5
-- go through the array of peers and collect acknowledgments
6
acks: *string
7
for peer <- peers par:
8
on peer:
9
acks <- Service.long_task()
10
11
-- if 10 acks collected or 1 second passed, return acks
12
join acks[9]
13
par Peer.timeout(1000, "timeout")
14
<- acks
Copied!
And here's how to approach error handling when using Peer.timeout
1
-- Peer.timeout comes from the standard library
2
import "@fluencelabs/aqua-lib/builtin"
3
​
4
func getOrNot() -> string:
5
status: *string
6
res: *string
7
-- Move execution to another peer
8
on "other peer":
9
res <- Srv.someFunction()
10
status <<- "ok"
11
-- In parallel with the previous on, run timeout on this peer
12
par status <- Peer.timeout(1000, "timeout")
13
14
-- status! waits for the first write to happen
15
if status! == "timeout":
16
-- Now we know that "other peer" was not able to respond within a second
17
-- Do some failover
18
res <<- "providing a local failover value"
19
20
<- res!
Copied!