Structure of Rusoto app with mutable shared state

help
#1

Hi there,

I would very much like to get suggestions regarding the architecture / approach for the following problem.

I'm trying to write a simple Rusoto utility to carry out rolling updates to AWS auto-scaling groups. The process goes as follows:

List all ASG and for each one that needs updating, run the following loop:

  • Launch a new instance;
  • Wait for the new instance to become ready (cold take a couple minutes);
  • Terminate an old instance;
  • Wait for the old instance to be removed from the group;
  • Loop until all instances have bounced.

My idea would be to have a single mutable shared state that would be refreshed from the AWS API every 10s. Then all updating loops would wait for the state to be refreshed, carry one piece of work, wait for a new state to be available, complete one more step, etc. So one writer N readers and some kind of synchronisation.

I have at the moment a state-updating loop that looks like this.

use async_trait::async_trait;
use tokio::time::delay_for;

// Hashmap providing a summary status for each ASG
pub struct AsgSummary {
  summary: Rc<RefCell<HashMap::<String, AsgStatus>>>
}

#[async_trait]
pub trait Summary {
  async fn update_summary(&self);
}

#[async_trait]
impl Summary for AsgSummary {
  async fn update_summary(&self) {
    loop {
      let mut summ = HashMap::<String, AsgStatus>::new();

      summ = <calculate a new summary>

      self.summary.replace(summ);

      delay_for(Duration::from_secs(10)).await;
    }
  }
}

#[tokio::main]
async fn main() {
  let asg = AsgSummary::new();
  asg.update_summary().await;
}

Now this piece of code doesn't compile due to the following error:

error: future cannot be sent between threads safely

    | |___^ future returned by `__update_summary` is not `Send`
    |
    = help: within `AsgSummary<'_>`, the trait `Sync` is not implemented for `RefCell<HashMap<String, AsgStatus>>`

Which makes sense in general but since the code will be mostly waiting for AWS to do the heavy lifting, a single-threaded approach should be perfectly ok.

Also I can't seem to figure out how to have the worker loops wait for a new state to become available.

Any help would be greatly appreciated!

Chris

#2

Use Arc<Mutex<...>> instead of Rc<RefCell<...>> as async code must typically be thread-safe. For more info, see the shared state chapter in the Tokio tutorial.

#3

As for why the error was triggered in this case, it is due to the async_trait crate, which requires that all of its methods are thread-safe by default. See the documentation for async-trait if you want to disable that.

#4

Hi Alice,

thanks a lot for the nudge, that seems to work indeed! (No more compilation errors at least :slight_smile: )

#[async_trait(?Send)]
pub trait Summary {
  async fn update_summary(&self);
}

Besides, would you have any recommendation as to how to sync the workers, i.e., let them block and be awaken when the writer updates the shared state?

Best regards,
Chris

#5

why not have them listen for a message on a channel? When the configurator updates the config, it can message all the workers to let them know work is ready.

You could even create a fresh config each round rather than mutating the old one, and send that over the channel, and then you could just have an Arc with no Mutex.

#6

Tokio defines a special kind of channel called a watch channel for this purpose.

1 Like
#7

Ok that looks like the correct approach. I will try to implement that.

Thanks a lot for the valuable help!
Chris