Mixing select! and async calls in Rust

Question

I am building a small app that's supposed to schedule two tasks (based on rusoto AWS SDK) at different intervals: every X seconds, run one task, and every Y seconds, run the other.

I found crate crossbeam that offers a tick timer and a select! macro and put them all together like this:

fn main() -> Result<(), Error> {
  let cloudwatch_client = rusoto_cloudwatch::CloudWatchClient::new();
  let rt = Runtime::new().unwrap();
  let tick_a = tick(Duration::from_secs(60));
  let tick_b = tick(Duration::from_secs(30));

  loop {
    select! {
      recv(tick_a) -> _ => {
        rt.block_on(cloudwatch_client.put_metric_data( /* ... */ ));
      },
      /* similar for tick_b */
    }
  }
}

This compiles, however the program panics with thread 'main' panicked at 'not currently running on the Tokio runtime.'. This seems to come from the Rusoto call, by analyzing the backtrace.

What am I missing here? Is there a way to make this work? Is there a better way to handle the scheduling of tasks at intervals in Rust?

Please note that this question does not seem to address my issue. That question starts out by using futures::executor::block_on function and is solved by using the block_on method implemented by the tokio Runtime. I am already using the block_on method in Runtime.

Answer 1

thread 'main' panicked at 'not currently running on the Tokio runtime.'

This error will show up if the version of the tokio runtime that is required by a particular library is not actively running - since each major version uses different thread-local variables and more than 1 major version of a library can be included in the same build.

In your case you might have a Tokio 0.3 runtime running, but rusoto expects a Tokio 0.2 runtime. When rusoto then tries to perform IO via Tokio 0.2 (which is also included in the build), that one detects that no runtime is active and yields the error.

To fix this, make sure to use just a single tokio version in your project. You likely need to downgrade tokio to 0.2 via Cargo.toml, since there might be no newer rusoto version available.

One more unrelated suggestion:

Instead of using crossbeam for timers, you can also run the "whole thing" inside a tokio runtime: You can use tokio::select! and tokio timers for doing what you do with crossbeam here.

See https://docs.rs/tokio/0.2.24/tokio/time/fn.interval.html and https://docs.rs/tokio/0.2.24/tokio/macro.select.html (which has examples similar to your use-case)

Answer 2

CloudWatchClient::put_metric_data returns a RusotoFuture which has a sync method does what you want. So instead of:

let cloudwatch_client = rusoto_cloudwatch::CloudWatchClient::new();
let rt = Runtime::new().unwrap();
rt.block_on(cloudwatch_client.put_metric_data(/* ... */));

You can do:

let cloudwatch_client = rusoto_cloudwatch::CloudWatchClient::new();
cloudwatch_client.put_metric_data(/* ... */).sync();

Which is also what the official docs recommend.

Answer 3

Please note:

Note that from v0.43.0 onward, Rusoto uses Rust's std::future::Future, and the Tokio 0.2 ecosystem. From v0.46.0 onward, Rusoto uses the Tokio 1.0 ecosystem.

I'd recommend the following dependencies for a simple S3 call:

[dependencies]  
rusoto_core = "0.46.0"                            
rusoto_s3 = "0.46.0"                                                    
tokio = {version = "1.0",  features = ["full"]}  

The call can be implemented like this:

use rusoto_core::Region;
use rusoto_s3::{S3, S3Client};

#[tokio::main]
async fn main() {
    let region = Region::UsEast1;
    let client = S3Client::new(region);

    match client.list_buckets().await {
        Ok(output) => match output.buckets {
            Some(bucket_list) => {
                println!("Buckets in S3:");

                for bucket in bucket_list {
                    println!("{:?}", bucket.name);
                }
            }
            None => println!("No buckets!"),
        },
        Err(error) => {
            println!("Error: {:?}", error);
        }
    }
}

Set credentials to avoid credentials error.