Render Jinja on a separate green thread

[liketoeatcheese]

Hi :) First of all, THANKYOU for your godly work!

I'm trying to achieve the following scenario, I need to do a server-side event scenario from a few different areas in parallel. E.g: reading 2 files with the same name from 2 different folders,

My thought process is spawning a tokio::task (thread) for each folder and render the template on each thread. I'm using AutoReload which impl<'reloader> ![Send] for [EnvironmentGuard]<'reloader>, so I'm not sure how would I achieve this. Any thoughts?

[mitsuhiko]

I would probably put the AutoReloader into some Arc, then clone it into the task and then acquire the guard within that.

[liketoeatcheese]

That's what I did, but I couldn't get it to work. I'm using Actix::web, so I did wrap the reloader to web::Data, which under the hood is an Arc. So, theoretically, I just need to clone it again whenever I want to chuck it in a thread (Even though it theoretically should work, but I also tried wrapping the whole thing to an Arc, but no luck). I try to isolate the issue into this small code:

main.rs

use actix_web::{get, web, App, Error as ActixError, HttpServer, Responder};
use actix_web_lab::sse;
use minijinja::context;
use minijinja::{path_loader, Environment};
use minijinja_autoreload::AutoReloader;
use std::path::PathBuf;
use std::time::Duration;
use std::{env, io};
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;

impl actix_web::error::ResponseError for CustomError {}

#[get("/events")]
async fn events(
    minijinja_env: web::Data<minijinja_autoreload::AutoReloader>,
) -> impl Responder {
    let (tx, rx) = mpsc::channel(3);
    let stream = ReceiverStream::new(rx);
    let tasks = vec![1, 2];

    for task_id in tasks {
        let minijinja_env = minijinja_env.clone();
        // let env = std::sync::Arc::new(minijinja_env.clone());

        let tx = tx.clone();
        tokio::spawn(async move {
            let mut counter = 1;
            loop {
                tokio::time::sleep(Duration::from_secs(4)).await;
                let ctx = context! {
                    task_id,
                    counter
                };
                let env = minijinja_env.acquire_env().unwrap();
                let test_html = env.get_template("test.html").unwrap().render(ctx).unwrap();

                let event = sse::Data::new(test_html).event(format!("task {}", task_id));
                if tx
                    .send(Ok::<_, io::Error>(sse::Event::Data(event)))
                    .await
                    .is_err()
                {
                    break;
                }
                counter += 1;
            }
        });
    }

    sse::Sse::from_stream(stream)
}

#[actix_web::main]
async fn main() -> io::Result<()> {
    // If DISABLE_AUTORELOAD is set, then the path tracking is disabled.
    let disable_autoreload = env::var("DISABLE_AUTORELOAD").as_deref() == Ok("1");

    // If FAST_AUTORELOAD is set, then fast reloading is enabled.
    let fast_autoreload = env::var("FAST_AUTORELOAD").as_deref() == Ok("1");

    // The closure is invoked every time the environment is outdated to
    // recreate it.
    let reloader = AutoReloader::new(move |notifier| {
        let template_path = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("templates");
        let mut env = Environment::new();
        env.set_loader(path_loader("templates"));

        if fast_autoreload {
            notifier.set_fast_reload(true);
        }

        // if watch_path is never called, no fs watcher is created
        if !disable_autoreload {
            notifier.watch_path(&template_path, true);
        }
        Ok(env)
    });
    println!("Starting server at http://localhost:8080");
    let jinja = web::Data::new(reloader);

    HttpServer::new(move || App::new().app_data(jinja.clone()).service(events))
        .bind(("127.0.0.1", 8080))?
        .run()
        .await
}

Cargo.toml

[package]
name = "stream_data"
version = "0.1.0"
edition = "2021"

[dependencies]
actix-web-lab = "0.20.2"
# Actix web related
actix-web = { version = "4.6.0", features = ["rustls-0_22"] }
minijinja = { version = "2.1.0", features = ["loader"] }
minijinja-autoreload = "2.1.0"
tokio = { version = "1.38.1", features = ["full"] }
tokio-stream = "0.1.15"

The error:

error: future cannot be sent between threads safely
   --> src/main.rs:46:9
    |
46  | /         tokio::spawn(async move {
47  | |             let mut counter = 1;
48  | |             loop {
49  | |                 tokio::time::sleep(Duration::from_secs(4)).await;
...   |
66  | |             }
67  | |         });
    | |__________^ future created by async block is not `Send`
    |
    = help: within `{async block@src/main.rs:46:22: 67:10}`, the trait `Send` is not implemented for `std::sync::MutexGuard<'_, Option<Environment<'static>>>`, which is required by `{async block@src/main.rs:46:22: 67:10}: Send`
note: future is not `Send` as this value is used across an await
   --> src/main.rs:60:22
    |
54  |                 let env = minijinja_env.acquire_env().unwrap();
    |                     --- has type `EnvironmentGuard<'_>` which is not `Send`
...
60  |                     .await
    |                      ^^^^^ await occurs here, with `env` maybe used later
note: required by a bound in `tokio::spawn`
   --> /home/wpham/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.38.1/src/task/spawn.rs:166:21
    |
164 |     pub fn spawn<F>(future: F) -> JoinHandle<F::Output>
    |            ----- required by a bound in this function
165 |     where
166 |         F: Future + Send + 'static,
    |                     ^^^^ required by this bound in `spawn`

[mitsuhiko]

1. Change AutoReloader::new(...) to Arc::new(AutoReloader::new(...))
2. Change web::Data<minijinja_autoreload::AutoReloader> to web::Data<Arc<minijinja_autoreload::AutoReloader>>
3. Ensure that env is not held across await points like so:

let test_html = {
    let env = minijinja_env.acquire_env().unwrap();
    env.get_template("test.html").unwrap().render(ctx).unwrap()
};

[liketoeatcheese]

Rightt!! Yea the key is this:

let test_html = {
    let env = minijinja_env.acquire_env().unwrap();
    env.get_template("test.html").unwrap().render(ctx).unwrap()
};

Don't need to create Arc::new(AutoReloader) because web::Data is already an Arc itself.

Seems like acquire_env() is like a lock. And it needs to be released while doing in parallel?

And thank you :) Appreciate it!