Intro

If you think about Transactions closely, they are pretty interesting beasts:

each shard knows the complete logic corresponding to each transaction, step-by-step, because they all share the same binary;
but generally no single shard can execute the transaction completely - from start to end, because the data is distributed across the entire database;
so we need a way to suspend execution on one shard and resume it on another one, without losing the data or replaying the transaction from the beginning.

This means, that we need something asynchronous, like futures, but with an ability to wake up on a different machine after each .await. Obviously, standard Rust futures do not support this kind of functionality. In fact, this is a very rare and specific functionality that is usually achieved by specialized libraries. After conducting a research, it turned out that the most promising primitive to achieve this kind of behaviour are coroutines.

If you’ve never heard of coroutines before, coroutines (fibers, green threads) are programs which may suspend their operations and then resume from the moment of previous suspension. They might seem similar to generators and they are. Both generators and coroutines emit results and suspend execution, until called again. The difference is that coroutines can accept new arguments from the outside at each suspension point:

sequenceDiagram
  participant Parent
  participant Coroutine

  Parent ->> Coroutine: call with args1
  
	activate Coroutine
  Coroutine ->> Coroutine: process args1
  Coroutine ->> Coroutine: persist state
	Coroutine ->> Parent: yield res1
	deactivate Coroutine

  Parent ->> Coroutine: call with args2

	activate Coroutine
  Coroutine ->> Coroutine: retrieve state
  Coroutine ->> Coroutine: process args2
  Coroutine ->> Coroutine: persist state
	Coroutine ->> Parent: yield res2
	deactivate Coroutine

  Parent ->> Coroutine: call with args3

	activate Coroutine
  Coroutine ->> Coroutine: retrieve state
  Coroutine ->> Coroutine: process args3
  Coroutine ->> Coroutine: destroy self
	Coroutine ->> Parent: return res3
	deactivate Coroutine

In order to pass coroutines between shards, we have to make them serializable. This is possible, but this is an even more rare feature across the ecosystem. For example, quasar’s fibers are serializable. In fact, there even exists a pretty well known DLT platform built on top of that library - Corda. This platform also provides a notion of sharded transactions, but in a different context, then union-db. And it also utilizes this property of fibers being serializable.

There is no similar library for Rust, which means that the only option for us is to build it ourselves. In order to do that, we need two new primitives available to our canisters:

the Coroutine trait, which will essentially transform any data type into a state machine;
the Dispatcher Service, which is a runtime that is responsible for Coroutine execution.

Coroutines

Take a look at this code:

#[coroutine]
fn example(a: u64, b: u64) -> u64 {
	let sum: u64 = a + b;
	
	let (result, cycles): (Vec<u8>, u128) = yield call(can_id, "do_example", sum, 0);
	let another_sum: u64 = decode(result).expect("invalid callback arg");

	if cycles > 0 {
		another_sum
	} else {
		sum
	}
}

This code will perform a math operation, then make an inter-canister call and then will submit the result to the original caller (another coroutine) with either the value it computed itself, or the value it received from another canister. Notice, there is no async/.await keywords, but the code works as there was an asynchronous call. The interesting part about this code is that it can be desugared into something like this:

fn example(a: u64, b: u64) -> (TaskId, ExampleCoroutine0) {
	let task_id = dispatcher().generate_task_id();

	(task_id, ExampleCoroutine0 { task_id, a, b })
}

struct ExampleCoroutine0 {
	task_id: TaskId, 
	a: u64, 
	b: u64,
}

struct ExampleCoroutine1 {
	task_id: TaskId, 
	sum: u64,
}

impl Coroutine for ExampleCoroutine0 {
	fn run(mut self, ev: Option<SysEvent>) {
		debug_assert!(ev.is_none());

		let sum: u64 = self.a + self.b;

		let task = Task::Coroutine(Box::new(ExampleCoroutine1 { task_id: self.task_id, sum }));
		
		let call_id = dispatcher().call(can_id, "do_example", sum, 0);
		dispatcher().dispatch_on(
			SysEventSelector::CallbackInvoked { call_id },
			task
		);
	}
}

impl<T> Coroutine for ExampleCoroutine1<T> {
	fn run(mut self, ev: Option<SysEvent>) {
		debug_assert!(ev.is_some());

		let (result, cycles): (Vec<u8>, u128) = unsafe { ev.unwrap_unchecked().unwrap_callback_invoked().body() };
		let another_sum: u64 = decode(result).expect("invalid callback arg");

		let __coroutine_result: u64 = if cycles > 0 {
			another_sum
		} else {
			self.sum
		};

		dispatcher().consume_event(SysEvent::TaskCompleted { 
			task_id: self.task_id, 
			body: TaskCompleteBody(encode(__coroutine_result).unwrap()) 
		});
	}
}

Table of contents

Intro

Coroutines