Clustered actors

Messages that travel

Move the device manager to a node of its own and send it readings and questions across the wire.

In this step the device manager of the getting started tutorial moves to a node of its own. The first node sends it readings and asks it for averages, with the same tell and ask as before. For that, its messages have to cross the wire.

Messages for the wire

The device manager, its groups, devices and queries are the ones of the getting started tutorial, with one change: names are String instead of &'static str. A message that arrives from another node is deserialized from bytes, so it cannot borrow a string that lives in the sender's program.

Only the messages that cross the wire need more. That is just Command, the messages for the device manager:

tellus/examples/tutorials/clustered/02_remotable_messages.rs

#[derive(Serialize, Deserialize)]
enum Command {
    Record {
        group_name: String,
        device_name: String,
        degrees: f64,
    },

    Retire {
        group_name: String,
        device_name: String,
    },

    Average {
        group_name: String,
        reply_to: ReplyTo<Option<f64>>,
    },
}

impl Manifest for Command {
    const MANIFEST: &'static str = "device-manager-command";
}

The serde derives make Command encodable. Average carries a ReplyTo, which is serializable too, so the reply finds its way back to the asking node. Manifest gives the message type a name that every node agrees on. A reference looked up by name needs it, so the receiving node can tell that the name means the same type there. The messages of the groups, devices and queries never leave the device manager's node, so they need neither serde nor a manifest.

The device manager node

The second node now runs the device manager:

tellus/examples/tutorials/clustered/02_remotable_messages.rs

async fn manager_node() -> anyhow::Result<()> {
    let (addr, seeds) = node_args()?;
    join_cluster(addr, seeds).await?;

    let system = ActorSystem::new(DeviceManager);
    cluster::register(&Key::new(MANAGER_KEY), system.root())
        .context("registering the device manager")?;
    system
        .terminated()
        .await
        .context("awaiting actor system termination")
}

After joining the cluster, it spawns the device manager as the root actor of its actor system and registers it with the receptionist under a Key. A key is a name together with the message type, here Key<Command>, so a lookup returns references of the right type. The registration is replicated to every other member.

Finding the device manager

The first node started the device manager's node, so it knows its address and looks the device manager up there:

tellus/examples/tutorials/clustered/02_remotable_messages.rs

async fn lookup_manager(addr: SocketAddr) -> anyhow::Result<ActorRef<Command>> {
    let key = Key::new(MANAGER_KEY);
    let lookup = async {
        loop {
            let managers =
                receptionist::lookup_at(&key, addr).context("looking up the device manager")?;
            match managers.into_iter().next() {
                Some(manager) => return anyhow::Ok(manager),
                None => sleep(LOOKUP_INTERVAL).await,
            }
        }
    };

    timeout(LOOKUP_TIMEOUT, lookup)
        .await
        .context("no device manager registered within the timeout")?
}

lookup_at answers from the registrations replicated to this node, without asking the other node. Right after bootstrap it answers empty, because the registration has not arrived yet. So lookup_manager retries, bounded by one timeout.

The first node

tellus/examples/tutorials/clustered/02_remotable_messages.rs

async fn first_node() -> anyhow::Result<()> {
    let seeds = reserved_addrs(2)?;
    let _manager_node = spawn_node(MANAGER_NODE_ARG, seeds[1], &seeds)?;
    join_cluster(seeds[0], seeds.clone()).await?;
    let manager = lookup_manager(seeds[1]).await?;

    let readings = [
        ("attic", "left", 21.5),
        ("attic", "right", 19.1),
        ("cellar", "boiler", 11.0),
        ("cellar", "boiler", f64::NAN),
    ];
    for (group_name, device_name, degrees) in readings {
        manager.tell(Command::Record {
            group_name: group_name.to_string(),
            device_name: device_name.to_string(),
            degrees,
        });
    }
    manager.tell(Command::Retire {
        group_name: "attic".to_string(),
        device_name: "right".to_string(),
    });

    for group_name in ["attic", "cellar", "garage"] {
        let average = manager
            .ask(ASK_TIMEOUT, |reply_to| Command::Average {
                group_name: group_name.to_string(),
                reply_to,
            })
            .await
            .context("asking for the average temperature")?;
        match average {
            Some(degrees) => println!("Average temperature in {group_name}: {degrees:.2}"),
            None => println!("No readings in {group_name}"),
        }
    }
    Ok(())
}

Apart from the lookup, this is the main of the getting started tutorial. The device manager is on another node now, but tell and ask are the same.

Run it

From a clone of the tellus repository:

cargo run --quiet -p tellus --features cluster-dev --example clustered_02_remotable_messages

It prints:

Average temperature in attic: 21.50
No readings in cellar
No readings in garage

The averages are those of the getting started tutorial. The boiler still fails on its invalid reading, and the device manager's node logs that to stderr.

What you learned

Messages that cross the wire derive serde, and those looked up under a key also implement Manifest. register and lookup_at connect two nodes through a key, and tell and ask work on a remote reference as on a local one. Next, the first node finds the device manager without knowing where it runs.