The coroutine mental model
In this chapter i want to show you what we know from traditional patterns, how it might look like when done using async logic as coroutines.
Do you remember boss fight state machines and the work we needed to do to run them manually to achieve sequential state flow?
Here is how all of that could be expressed as async action timeline:
use crossterm::{
event::{Event, KeyCode},
style::Stylize,
};
use moirai::{
coroutine::{CompletionImportance, meta, with_importance},
job::{JobHandle, JobLocation},
third_party::intuicio_data::{managed::DynamicManagedLazy, shared::AsyncShared},
};
use moirai_book_samples::{
coroutines::{Coroutines, next_frame},
game::{Game, GameState, GameStateChange},
terminal::Terminal,
utils::is_key_pressed,
};
use std::time::Duration;
fn main() {
Game::new(Example::default()).run_blocking();
}
const TIME_SCALE: f32 = 5.0;
const DAMAGE_STAMINA_MIN_REQUIREMENT: f32 = 5.0;
const LOG_CAPACITY: usize = 5;
const EVENTS_META: &str = "~events~";
const DELTA_TIME_META: &str = "~delta-time~";
const LOG_META: &str = "~log~";
struct Example {
terminal: Terminal,
coroutines: Coroutines,
player: AsyncShared<Character>,
enemy: AsyncShared<Character>,
log: Vec<String>,
}
impl Default for Example {
fn default() -> Self {
let coroutines = Coroutines::default();
let player = AsyncShared::new(Character::new("Player"));
let enemy = AsyncShared::new(Character::new("Enemy"));
let player_shared = player.clone();
let enemy_shared = enemy.clone();
player.write().unwrap().timeline(&coroutines, async move {
loop {
let events = events().await;
if is_key_pressed(&events, KeyCode::Enter) {
attack(&player_shared, &enemy_shared).await;
} else if is_key_pressed(&events, KeyCode::Char(' ')) {
block(player_shared.clone()).await;
} else {
update(&player_shared).await;
}
next_frame().await;
}
});
let player_shared = player.clone();
let enemy_shared = enemy.clone();
enemy.write().unwrap().timeline(&coroutines, async move {
loop {
charge(&enemy_shared, 10.0).await;
attack(&enemy_shared, &player_shared).await;
charge(&enemy_shared, 10.0).await;
block(enemy_shared.clone()).await;
charge(&enemy_shared, 20.0).await;
attack(&enemy_shared, &player_shared).await;
}
});
Self {
terminal: Terminal::default(),
coroutines,
player,
enemy,
log: Vec::new(),
}
}
}
impl GameState for Example {
fn frame(&mut self, mut delta_time: Duration) -> GameStateChange {
let mut events = Terminal::events().collect::<Vec<_>>();
draw_system(&mut self.terminal, &self.player, &self.enemy, &self.log);
{
let (events_lazy, _events_lifemtime) = DynamicManagedLazy::make(&mut events);
let (dt_lazy, _dt_lifemtime) = DynamicManagedLazy::make(&mut delta_time);
let (log_lazy, _log_lifetime) = DynamicManagedLazy::make(&mut self.log);
self.coroutines.run_frame(
[
(EVENTS_META.into(), events_lazy.into()),
(DELTA_TIME_META.into(), dt_lazy.into()),
(LOG_META.into(), log_lazy.into()),
]
.into_iter()
.collect(),
);
}
{
let player = self.player.write().unwrap();
if !player.is_alive() {
player.timeline.cancel();
}
let enemy = self.enemy.write().unwrap();
if !enemy.is_alive() {
enemy.timeline.cancel();
}
}
if is_key_pressed(&events, KeyCode::Esc) {
GameStateChange::Quit
} else {
GameStateChange::None
}
}
}
struct Character {
name: String,
health: f32,
stamina: f32,
timeline: JobHandle<()>,
}
impl Character {
fn new(name: impl ToString) -> Self {
Self {
name: name.to_string(),
health: 100.0,
stamina: 0.0,
timeline: Default::default(),
}
}
fn timeline(
&mut self,
coroutines: &Coroutines,
timeline: impl Future<Output = ()> + Send + Sync + 'static,
) {
self.timeline = coroutines
.queue()
.spawn(JobLocation::Local, timeline)
.cancel_on_drop();
}
fn is_alive(&self) -> bool {
self.health > 0.0
}
}
async fn update(this: &AsyncShared<Character>) {
let dt = delta_time().await.as_secs_f32() * TIME_SCALE;
let mut this = this.write().unwrap();
this.stamina = (this.stamina + dt).min(100.0);
}
async fn attack(this: &AsyncShared<Character>, target: &AsyncShared<Character>) {
log(format!(
"{} attacks {} with {:.02} hitpoints",
this.read().unwrap().name,
target.read().unwrap().name,
this.read().unwrap().stamina
))
.await;
let mut this = this.write().unwrap();
let mut target = target.write().unwrap();
if this.stamina >= DAMAGE_STAMINA_MIN_REQUIREMENT {
target.health = (target.health - this.stamina).max(0.0);
this.stamina = 0.0;
}
}
async fn charge(this: &AsyncShared<Character>, target_stamina: f32) {
loop {
let dt = delta_time().await.as_secs_f32() * TIME_SCALE;
{
let mut this = this.write().unwrap();
this.stamina = (this.stamina + dt).min(100.0);
if this.stamina >= target_stamina {
break;
}
}
next_frame().await;
}
}
async fn block(this: AsyncShared<Character>) {
log(format!(
"{} blocks with {:.02} cooldown",
this.read().unwrap().name,
this.read().unwrap().stamina
))
.await;
let health = this.read().unwrap().health;
loop {
let dt = delta_time().await.as_secs_f32() * TIME_SCALE;
{
let mut this = this.write().unwrap();
this.health = health;
this.stamina = (this.stamina - dt).min(100.0);
if this.stamina <= 0.0 {
break;
}
}
next_frame().await;
}
log(format!("{} stops blocking", this.read().unwrap().name)).await;
}
async fn parry(this: AsyncShared<Character>, target: AsyncShared<Character>, mut cooldown: f32) {
let origin_health = this.read().unwrap().health;
let damage = loop {
let dt = delta_time().await.as_secs_f32() * TIME_SCALE;
cooldown -= dt;
if cooldown <= 0.0 {
return;
}
let current_health = this.read().unwrap().health;
if current_health < origin_health {
break origin_health - current_health;
}
next_frame().await;
};
log(format!(
"{} parries {} with {:.02} hitpoints",
this.read().unwrap().name,
target.read().unwrap().name,
damage
))
.await;
let mut this = this.write().unwrap();
let mut target = target.write().unwrap();
target.health = (target.health - damage).max(0.0);
this.stamina = 0.0;
this.health = (this.health + damage).min(100.0);
}
fn draw_system(
terminal: &mut Terminal,
player: &AsyncShared<Character>,
enemy: &AsyncShared<Character>,
log: &[String],
) {
terminal.begin_draw(true);
let player = player.read().unwrap();
let enemy = enemy.read().unwrap();
match (player.is_alive(), enemy.is_alive()) {
(true, true) => {
terminal.display([1, 1], "Player:".green().bold());
terminal.display([2, 2], format!("Health: {:.1}", player.health));
terminal.display([2, 3], format!("Stamina: {:.1}", player.stamina));
terminal.display([1, 6], "Enemy:".red().bold());
terminal.display([2, 7], format!("Health: {:.1}", enemy.health));
terminal.display([2, 8], format!("Stamina: {:.1}", enemy.stamina));
terminal.display([20, 1], "Log:".underlined());
for (i, log) in log.iter().rev().take(LOG_CAPACITY).enumerate() {
terminal.display([21, 2 + i as u16], log);
}
}
(false, false) => {
terminal.display([1, 1], "It's a draw!".yellow().bold());
}
(true, false) => {
terminal.display([1, 1], "You are victorious!".green().bold());
terminal.display([2, 2], format!("Health: {:.1}", player.health));
}
(false, true) => {
terminal.display([1, 1], "You have been defeated!".red().bold());
terminal.display([2, 2], format!("Health: {:.1}", enemy.health));
}
}
terminal.end_draw();
}
async fn delta_time() -> Duration {
*meta::<Duration>(DELTA_TIME_META)
.await
.unwrap()
.read()
.unwrap()
}
async fn events() -> Vec<Event> {
meta::<Vec<Event>>(EVENTS_META)
.await
.unwrap()
.read()
.unwrap()
.clone()
}
async fn log(content: impl ToString) {
meta::<Vec<String>>(LOG_META)
.await
.unwrap()
.write()
.unwrap()
.push(content.to_string());
}
// ===
#[allow(dead_code)]
async fn example_with_parry(
enemy_shared: AsyncShared<Character>,
player_shared: AsyncShared<Character>,
) {
loop {
charge(&enemy_shared, 10.0).await;
attack(&enemy_shared, &player_shared).await;
charge(&enemy_shared, 10.0).await;
let parry = parry(enemy_shared.clone(), player_shared.clone(), 10.0);
let block = block(enemy_shared.clone());
with_importance(vec![
CompletionImportance::ignored(parry),
CompletionImportance::required(block),
])
.await;
charge(&enemy_shared, 20.0).await;
attack(&enemy_shared, &player_shared).await;
}
}
#[allow(dead_code)]
async fn example_with_waiting_for_input(
player: &AsyncShared<Character>,
enemy: &AsyncShared<Character>,
) {
loop {
let events = events().await;
if is_key_pressed(&events, KeyCode::Enter) {
attack(player, enemy).await;
} else if is_key_pressed(&events, KeyCode::Char(' ')) {
block(player.clone()).await;
} else {
update(player).await;
}
}
}Action timelines (Coroutines)
Action timelines, or coroutines as game developers tends to call them, are used to describe async logic flow in more sequential (to be more precise: procedural) way, reducing suspension and continuation boundary to single place (.await), instead of traditional reactive, scattered and manual state machine management.
It’s worth noting that neihter of those is better than the other - they both express same end goal, but in different ways, and we shouldn’t feel guilty for using one over another.
Coroutines allow constructing such state machines in more procedural way in code, automatically, as suspension and continuation is a first class concept in async, which covers perfectly the declaration and execution part of work that’s spread across time, but not across codebase - a state machine that we don’t need to handle manually, built for us by the compiler from single code block.
Automatic state machines
Imagine a scenario where we start with those 3 states we have made for NPC fight pattern, then we want to add a state where NPC can parry incoming player attack while it’s in a blocking phase.
Async can provide us with primitives that allow us to easily express “while doing this, you can also try do that other thing” without much of an architectural refactoring, as we would might have need to do with manual state machines.
use crossterm::{
event::{Event, KeyCode},
style::Stylize,
};
use moirai::{
coroutine::{CompletionImportance, meta, with_importance},
job::{JobHandle, JobLocation},
third_party::intuicio_data::{managed::DynamicManagedLazy, shared::AsyncShared},
};
use moirai_book_samples::{
coroutines::{Coroutines, next_frame},
game::{Game, GameState, GameStateChange},
terminal::Terminal,
utils::is_key_pressed,
};
use std::time::Duration;
fn main() {
Game::new(Example::default()).run_blocking();
}
const TIME_SCALE: f32 = 5.0;
const DAMAGE_STAMINA_MIN_REQUIREMENT: f32 = 5.0;
const LOG_CAPACITY: usize = 5;
const EVENTS_META: &str = "~events~";
const DELTA_TIME_META: &str = "~delta-time~";
const LOG_META: &str = "~log~";
struct Example {
terminal: Terminal,
coroutines: Coroutines,
player: AsyncShared<Character>,
enemy: AsyncShared<Character>,
log: Vec<String>,
}
impl Default for Example {
fn default() -> Self {
let coroutines = Coroutines::default();
let player = AsyncShared::new(Character::new("Player"));
let enemy = AsyncShared::new(Character::new("Enemy"));
let player_shared = player.clone();
let enemy_shared = enemy.clone();
player.write().unwrap().timeline(&coroutines, async move {
loop {
let events = events().await;
if is_key_pressed(&events, KeyCode::Enter) {
attack(&player_shared, &enemy_shared).await;
} else if is_key_pressed(&events, KeyCode::Char(' ')) {
block(player_shared.clone()).await;
} else {
update(&player_shared).await;
}
next_frame().await;
}
});
let player_shared = player.clone();
let enemy_shared = enemy.clone();
enemy.write().unwrap().timeline(&coroutines, async move {
loop {
charge(&enemy_shared, 10.0).await;
attack(&enemy_shared, &player_shared).await;
charge(&enemy_shared, 10.0).await;
block(enemy_shared.clone()).await;
charge(&enemy_shared, 20.0).await;
attack(&enemy_shared, &player_shared).await;
}
});
Self {
terminal: Terminal::default(),
coroutines,
player,
enemy,
log: Vec::new(),
}
}
}
impl GameState for Example {
fn frame(&mut self, mut delta_time: Duration) -> GameStateChange {
let mut events = Terminal::events().collect::<Vec<_>>();
draw_system(&mut self.terminal, &self.player, &self.enemy, &self.log);
{
let (events_lazy, _events_lifemtime) = DynamicManagedLazy::make(&mut events);
let (dt_lazy, _dt_lifemtime) = DynamicManagedLazy::make(&mut delta_time);
let (log_lazy, _log_lifetime) = DynamicManagedLazy::make(&mut self.log);
self.coroutines.run_frame(
[
(EVENTS_META.into(), events_lazy.into()),
(DELTA_TIME_META.into(), dt_lazy.into()),
(LOG_META.into(), log_lazy.into()),
]
.into_iter()
.collect(),
);
}
{
let player = self.player.write().unwrap();
if !player.is_alive() {
player.timeline.cancel();
}
let enemy = self.enemy.write().unwrap();
if !enemy.is_alive() {
enemy.timeline.cancel();
}
}
if is_key_pressed(&events, KeyCode::Esc) {
GameStateChange::Quit
} else {
GameStateChange::None
}
}
}
struct Character {
name: String,
health: f32,
stamina: f32,
timeline: JobHandle<()>,
}
impl Character {
fn new(name: impl ToString) -> Self {
Self {
name: name.to_string(),
health: 100.0,
stamina: 0.0,
timeline: Default::default(),
}
}
fn timeline(
&mut self,
coroutines: &Coroutines,
timeline: impl Future<Output = ()> + Send + Sync + 'static,
) {
self.timeline = coroutines
.queue()
.spawn(JobLocation::Local, timeline)
.cancel_on_drop();
}
fn is_alive(&self) -> bool {
self.health > 0.0
}
}
async fn update(this: &AsyncShared<Character>) {
let dt = delta_time().await.as_secs_f32() * TIME_SCALE;
let mut this = this.write().unwrap();
this.stamina = (this.stamina + dt).min(100.0);
}
async fn attack(this: &AsyncShared<Character>, target: &AsyncShared<Character>) {
log(format!(
"{} attacks {} with {:.02} hitpoints",
this.read().unwrap().name,
target.read().unwrap().name,
this.read().unwrap().stamina
))
.await;
let mut this = this.write().unwrap();
let mut target = target.write().unwrap();
if this.stamina >= DAMAGE_STAMINA_MIN_REQUIREMENT {
target.health = (target.health - this.stamina).max(0.0);
this.stamina = 0.0;
}
}
async fn charge(this: &AsyncShared<Character>, target_stamina: f32) {
loop {
let dt = delta_time().await.as_secs_f32() * TIME_SCALE;
{
let mut this = this.write().unwrap();
this.stamina = (this.stamina + dt).min(100.0);
if this.stamina >= target_stamina {
break;
}
}
next_frame().await;
}
}
async fn block(this: AsyncShared<Character>) {
log(format!(
"{} blocks with {:.02} cooldown",
this.read().unwrap().name,
this.read().unwrap().stamina
))
.await;
let health = this.read().unwrap().health;
loop {
let dt = delta_time().await.as_secs_f32() * TIME_SCALE;
{
let mut this = this.write().unwrap();
this.health = health;
this.stamina = (this.stamina - dt).min(100.0);
if this.stamina <= 0.0 {
break;
}
}
next_frame().await;
}
log(format!("{} stops blocking", this.read().unwrap().name)).await;
}
async fn parry(this: AsyncShared<Character>, target: AsyncShared<Character>, mut cooldown: f32) {
let origin_health = this.read().unwrap().health;
let damage = loop {
let dt = delta_time().await.as_secs_f32() * TIME_SCALE;
cooldown -= dt;
if cooldown <= 0.0 {
return;
}
let current_health = this.read().unwrap().health;
if current_health < origin_health {
break origin_health - current_health;
}
next_frame().await;
};
log(format!(
"{} parries {} with {:.02} hitpoints",
this.read().unwrap().name,
target.read().unwrap().name,
damage
))
.await;
let mut this = this.write().unwrap();
let mut target = target.write().unwrap();
target.health = (target.health - damage).max(0.0);
this.stamina = 0.0;
this.health = (this.health + damage).min(100.0);
}
fn draw_system(
terminal: &mut Terminal,
player: &AsyncShared<Character>,
enemy: &AsyncShared<Character>,
log: &[String],
) {
terminal.begin_draw(true);
let player = player.read().unwrap();
let enemy = enemy.read().unwrap();
match (player.is_alive(), enemy.is_alive()) {
(true, true) => {
terminal.display([1, 1], "Player:".green().bold());
terminal.display([2, 2], format!("Health: {:.1}", player.health));
terminal.display([2, 3], format!("Stamina: {:.1}", player.stamina));
terminal.display([1, 6], "Enemy:".red().bold());
terminal.display([2, 7], format!("Health: {:.1}", enemy.health));
terminal.display([2, 8], format!("Stamina: {:.1}", enemy.stamina));
terminal.display([20, 1], "Log:".underlined());
for (i, log) in log.iter().rev().take(LOG_CAPACITY).enumerate() {
terminal.display([21, 2 + i as u16], log);
}
}
(false, false) => {
terminal.display([1, 1], "It's a draw!".yellow().bold());
}
(true, false) => {
terminal.display([1, 1], "You are victorious!".green().bold());
terminal.display([2, 2], format!("Health: {:.1}", player.health));
}
(false, true) => {
terminal.display([1, 1], "You have been defeated!".red().bold());
terminal.display([2, 2], format!("Health: {:.1}", enemy.health));
}
}
terminal.end_draw();
}
async fn delta_time() -> Duration {
*meta::<Duration>(DELTA_TIME_META)
.await
.unwrap()
.read()
.unwrap()
}
async fn events() -> Vec<Event> {
meta::<Vec<Event>>(EVENTS_META)
.await
.unwrap()
.read()
.unwrap()
.clone()
}
async fn log(content: impl ToString) {
meta::<Vec<String>>(LOG_META)
.await
.unwrap()
.write()
.unwrap()
.push(content.to_string());
}
// ===
#[allow(dead_code)]
async fn example_with_parry(
enemy_shared: AsyncShared<Character>,
player_shared: AsyncShared<Character>,
) {
loop {
charge(&enemy_shared, 10.0).await;
attack(&enemy_shared, &player_shared).await;
charge(&enemy_shared, 10.0).await;
let parry = parry(enemy_shared.clone(), player_shared.clone(), 10.0);
let block = block(enemy_shared.clone());
with_importance(vec![
CompletionImportance::ignored(parry),
CompletionImportance::required(block),
])
.await;
charge(&enemy_shared, 20.0).await;
attack(&enemy_shared, &player_shared).await;
}
}
#[allow(dead_code)]
async fn example_with_waiting_for_input(
player: &AsyncShared<Character>,
enemy: &AsyncShared<Character>,
) {
loop {
let events = events().await;
if is_key_pressed(&events, KeyCode::Enter) {
attack(player, enemy).await;
} else if is_key_pressed(&events, KeyCode::Char(' ')) {
block(player.clone()).await;
} else {
update(player).await;
}
}
}Instead of just blocking, the enemy will try to parry the player’s upcoming attack while it’s in blocking phase of a timeline.
To help us with ergonomic expression of this, we use here an async primitive to run both states at the same time, where blocking state is required to complete, but parry is optional and can be terminated when blocking state completes.
So, as you can see, we didn’t had to spend much time on plugging in yet another state, as we did it just in one place, still keeping mental picture of how it fits within general states flow.
Awaiting game signals
Do you also remember traditional event-based logic flow?
Wanna see how easy would it get to express such construct with coroutines?
Here, let’s consider a scenario of quest timeline, where we expect player to go to a couple of places and pick up some stuff:
use crossterm::event::KeyCode;
use moirai::job::{JobHandle, JobLocation};
use moirai_book_samples::{
coroutines::{Coroutines, next_frame},
events::oneshot::{Receiver, Sender, channel},
game::{Game, GameState, GameStateChange},
terminal::Terminal,
utils::is_key_pressed,
};
use std::time::Duration;
fn main() {
Game::new(Example::default()).run_blocking();
}
struct Example {
_terminal: Terminal,
coroutines: Coroutines,
_quest: JobHandle<()>,
go_to_events: Sender<String>,
pick_up_object: Sender<String>,
quest_done: Receiver<()>,
}
impl Default for Example {
fn default() -> Self {
let coroutines = Coroutines::default();
let (go_to_events_sender, go_to_events_receiver) = channel();
let (pick_up_object_sender, pick_up_object_receiver) = channel();
let (quest_done_sender, quest_done_receiver) = channel();
let quest = coroutines
.queue()
.spawn(JobLocation::Local, async move {
println!("Starting quest...");
wait_for_event("tavern", &go_to_events_receiver).await;
println!("Arrived at the tavern.");
wait_for_event("food", &pick_up_object_receiver).await;
println!("Picked up food.");
wait_for_event("castle", &go_to_events_receiver).await;
println!("Arrived at the castle.");
wait_for_event("drink", &pick_up_object_receiver).await;
println!("Picked up drink.");
quest_done_sender.send(());
println!("Quest completed!");
})
.cancel_on_drop();
Self {
_terminal: Terminal::default(),
coroutines,
_quest: quest,
go_to_events: go_to_events_sender,
pick_up_object: pick_up_object_sender,
quest_done: quest_done_receiver,
}
}
}
impl GameState for Example {
fn frame(&mut self, _delta_time: Duration) -> GameStateChange {
let events = Terminal::events().collect::<Vec<_>>();
if is_key_pressed(&events, KeyCode::Char('q')) {
self.go_to_events.send("tavern".to_string());
println!("> go to tavern.");
}
if is_key_pressed(&events, KeyCode::Char('a')) {
self.go_to_events.send("castle".to_string());
println!("> go to castle.");
}
if is_key_pressed(&events, KeyCode::Char('w')) {
self.pick_up_object.send("food".to_string());
println!("> pick up food.");
}
if is_key_pressed(&events, KeyCode::Char('s')) {
self.pick_up_object.send("drink".to_string());
println!("> pick up drink.");
}
self.coroutines.run_frame(Default::default());
if self.quest_done.try_recv().is_some() {
println!("All done! Press ESC to quit.");
}
if is_key_pressed(&events, KeyCode::Esc) {
GameStateChange::Quit
} else {
GameStateChange::None
}
}
}
async fn wait_for_event(id: &str, events: &Receiver<String>) {
loop {
if let Some(event) = events.try_recv()
&& event == id
{
return;
}
next_frame().await;
}
}All we had to do was to declare waiting for certain things to happen in game, and so quest suspends until each step is completed. Being able to declare awaiting for certain game system signal as part of a bigger sequential timeline is quite a benefit!
Of course in Big Games ™, we prefer quests systems to be rather asset-driven, but quests are best at showing in general how to make action timelines awaiting on various game systems triggering events.