Example of how to use with arguments?

[bburdette]

I have an existing async function that takes four arguments:

pub async fn sync(
  dbpath: &Path,
  file_path: &Path,
  uid: i64,
  callbacks: &mut Callbacks,
) -> Result<PrivateReplyMessage, Box<dyn std::error::Error>> {

I'd like to run this in a job. Currently async closures are nightly, so I'm looking at making a function that takes the above arguments and returns another function that takes Monitor, runs sync, and sets the Job Status to 'sync complete' or whatever. The types and whatnot are a bit tricky though, would be nice to have an example.

[j-tai]

Good point, I should probably add some examples.

It looks like you have a bunch of borrowed data that the job needs. Since the job runs "in the background", you need to ensure they are either 'static, or wrap them in Arc or similar. (Or you can give ownership of the values to the job.)

Perhaps you're looking for something like this:

let dbpath: PathBuf = todo!();
let file_path: PathBuf = todo!();
let uid: i64 = todo!();
let callbacks: Callbacks = todo!();

let job = Job::start(move |mon| async move {
    writeln!(mon, "starting!");
    sync(&dbpath, &file_path, uid, &mut callbacks).await.unwrap();
    writeln!(mon, "done!");
});

Let me know if you have any other questions!

[bburdette]

Wow quick response! I gave that a go:

    PrivateRequests::SyncRemote => {
      let dbpath: PathBuf = config.orgauth_config.db.to_path_buf();
      let file_path: PathBuf = config.file_path.to_path_buf();
      let uid: i64 = uid;
      let job = Job::start(move |mon| async move {
        writeln!(mon, "starting!");
        sync::sync(&dbpath, &file_path, uid).await.unwrap();
        writeln!(mon, "done!");
      });

      Ok(PrivateReplyMessage {
        what: PrivateReplies::SyncStarted,
        content: serde_json::to_value("meh".to_string())?,
      })
    }

Because Callbacks is not Send, it was causing errors. For now I just moved that bit into the sync() function (didn't help 😄). Unfortunately that reveals that the sync function is chock-full of less tractable lack-of-Send errors in addition to the Callbacks, notably these third party lib issues:

• awc::Client::new()
• tokio_util::io::StreamReader;

These I don't think I can do without, or change! I don't suppose there's any cure for this lack of Send?

error.txt

[bburdette]

Looks like this is due to tokio::spawn requiring Send, and the requirement is passed on up. The nature of async thread pooling? Maybe I'll just spawn regular threads for this.

[j-tai]

Yeah, that's the nature of async runtimes -- there's no guarantee which thread will run the task, and tasks can even switch threads at any await point.

You could consider using tokio::task::spawn_blocking to run the code in a blocking thread:

let job = Job::start(move |mon| async move {
    writeln!(mon, "starting!");
    tokio::task::spawn_blocking(move || {
        // do your non-Send things here...
    })
    .await;
    writeln!(mon, "done!");
});

[bburdette]

Yeah that compiles. Even can pass in the monitor. Thx for the help!!

[j-tai]

Glad to help! :)

[bburdette]

Well I got around to trying the spawn_blocking solution, and while it did compile, it was just doing the thing where you get a future back which is never executed, lol. Adding an .await produces an error. It seems spawn_blocking gives you a thread to execute non-async, non-Send code, but my function is async and non-Send.

For now I resorted to just starting a regular thread and creating a new runtime and localset there. It runs!

      let job = Job::start(move |mon| async move {
          std::thread::spawn(move || {
            let rt = Runtime::new().unwrap();
            let local = LocalSet::new();
            let mut callbacks = &mut zknotes_callbacks();
            let r = local.block_on(&rt, sync::sync(&dbpath, &file_path, uid, &mut callbacks));
          });
        });

I did find this other approach in the tokio docs, where they suggest creating another thread with its own persistent localset and runtime. Then you send tasks to it with mpsc. I guess the rationale is to save creating and destroying the runtime and localset each time you want to run a task.

https://docs.rs/tokio/latest/tokio/task/struct.LocalSet.html#use-inside-tokiospawn

This seems very like the kind of thing one would find in a job manager! ; )

[j-tai]

Right, I suppose that makes sense. Looking into it a bit more, tokio requires Futures to be Send, and some of your libraries (awc) are incompatible with this assumption. I guess you should be using actix_rt instead, which allows !Send futures.

Right now, girlboss uses Tokio to spawn its tasks, which means it requires Send. Perhaps I could refactor my library to support multiple runtimes instead of just Tokio.

[bburdette]

That explains why I got this far, as the job has been running in an actix_web handler up to now which presumably uses actix_rt.

In the multiple runtime scenario one could make a Girlboss instance with actix_rt for !Send tasks, and another instance with tokio for Send stuff w extra efficiency. That sounds cool!