Getting started

Asking all devices

Collect answers from many actors with a short-lived query actor, without timeouts.

What is the average temperature of a group? To answer, the group has to ask all of its devices and combine their readings. Meanwhile it should keep handling other messages, and a device might retire before it answers. In this step a short-lived query actor does the asking.

The handover from the previous step has served its purpose, so devices now simply stop on Retire. The group's Read gives way to Average. Devices keep their Read: the query uses it.

One query actor per request

tellus/examples/tutorials/getting_started/05_query.rs

Incoming::Message(GroupMessage::Average(reply_to)) => {
    let query = context.spawn(Query);
    query.tell(QueryMessage::Start {
        devices: devices.values().cloned().collect(),
        reply_to,
    });
    Ok(Control::Continue(devices))
}

For every Average request, the group spawns a Query and gives it the current devices and the request's ReplyTo. Then the group is free for the next message. The query does the rest:

tellus/examples/tutorials/getting_started/05_query.rs

struct Query;

impl Actor for Query {
    type Message = QueryMessage;
    type State = QueryState;
    type Error = Infallible;

    fn init(&self, _: &ActorContext<Self::Message>) -> Result<Self::State, Self::Error> {
        Ok(QueryState::Starting)
    }

    fn receive(
        &self,
        context: &ActorContext<Self::Message>,
        incoming: Incoming<Self::Message>,
        state: Self::State,
    ) -> Result<Control<Self::State>, Self::Error> {
        match (incoming, state) {
            (Incoming::Message(QueryMessage::Start { devices, reply_to }), _) => {
                let mut pending = HashMap::new();
                for device in devices {
                    context.watch(&device);
                    let device_id = device.actor_id();
                    device.tell(DeviceMessage::Read(context.reply_to(move |reading| {
                        QueryMessage::Reading { device_id, reading }
                    })));
                    pending.insert(device_id, device);
                }
                Query::conclude(pending, 0.0, 0, reply_to)
            }

            (
                Incoming::Message(QueryMessage::Reading { device_id, reading }),
                QueryState::Running {
                    mut pending,
                    mut sum,
                    mut count,
                    reply_to,
                },
            ) => {
                if let Some(device) = pending.remove(&device_id) {
                    context.unwatch(&device);
                }
                if let Some(degrees) = reading {
                    sum += degrees;
                    count += 1;
                }
                Query::conclude(pending, sum, count, reply_to)
            }

            (
                Incoming::Terminated(device_id),
                QueryState::Running {
                    mut pending,
                    sum,
                    count,
                    reply_to,
                },
            ) => {
                // The ordering guarantee proves that this device's reading can no longer arrive.
                pending.remove(&device_id);
                Query::conclude(pending, sum, count, reply_to)
            }

            (_, QueryState::Starting) => unreachable!("the start message comes first"),
        }
    }
}

impl Query {
    fn conclude(
        pending: HashMap<ActorId, ActorRef<DeviceMessage>>,
        sum: f64,
        count: u32,
        reply_to: ReplyTo<Option<f64>>,
    ) -> Result<Control<QueryState>, Infallible> {
        if pending.is_empty() {
            reply_to.reply((count > 0).then(|| sum / f64::from(count)));
            Ok(Control::Stop)
        } else {
            Ok(Control::Continue(QueryState::Running {
                pending,
                sum,
                count,
                reply_to,
            }))
        }
    }
}

enum QueryMessage {
    Start {
        devices: Vec<ActorRef<DeviceMessage>>,
        reply_to: ReplyTo<Option<f64>>,
    },

    Reading {
        device_id: ActorId,
        reading: Option<f64>,
    },
}

enum QueryState {
    Starting,

    Running {
        pending: HashMap<ActorId, ActorRef<DeviceMessage>>,
        sum: f64,
        count: u32,
        reply_to: ReplyTo<Option<f64>>,
    },
}

The query's state is a small state machine. On Start it watches every device, sends each a Read and remembers it as pending. Each reading arrives through context.reply_to as a Reading message and is added to the sum. When no device is pending anymore, the query replies and stops.

Why the query cannot hang

A device may terminate before it answers. Then its Read is never handled, and without further care the query would wait forever.

Because the query watches every device, it receives Terminated for such a device. The ordering guarantee from the previous step tells the query that this device's reading can no longer arrive, so it stops waiting for it. No timeout is needed inside the actors. The only deadline is the one on ask in main, at the boundary to the outside world.

When a reading does arrive, the query calls context.unwatch. After unwatch returns, the query never receives a terminated signal for that device, even if one was already on its way.

Run it

tellus/examples/tutorials/getting_started/05_query.rs

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let system = ActorSystem::new(Group);
    let group = system.root();

    for (device_name, degrees) in [("left", 21.5), ("middle", 20.4), ("right", 19.1)] {
        group.tell(GroupMessage::Record {
            device_name,
            degrees,
        });
    }
    group.tell(GroupMessage::Retire {
        device_name: "right",
    });

    let average = group
        .ask(Duration::from_secs(1), GroupMessage::Average)
        .await
        .context("asking for the average temperature")?;
    match average {
        Some(degrees) => println!("Average temperature: {degrees:.2}"),
        None => println!("No readings"),
    }

    system.stop();
    system
        .terminated()
        .await
        .context("awaiting actor system termination")
}

cargo run --quiet -p tellus --example getting_started_05_query

Average temperature: 20.95

The device right retires right before the request, so the average covers left and middle only.

What you learned

A short-lived actor per request keeps the group responsive. Watching the actors you wait for, and relying on the ordering guarantee, replaces timeouts inside your actors. Next, devices fail.