A child thread borrowing `var`: [1, 2, 3]
A child thread borrowing `var2`: [3, 2, 1]
To clarify, it does work. The reason I am asking is because of the argument to s.spawn's closure is a scope object that can be used to spawn nested threads. But if I use the nested s.spawn() instead of creating another thread::scope I can't access var2:
use crossbeam_utils::thread;
fn main() {
let var = vec![1, 2, 3];
thread::scope(|s| {
s.spawn(|s| {
// Var2 can't be referenced inside of the closure below
let var2 = vec![3,2,1];
s.spawn(|_| {
println!("A child thread borrowing `var`: {:?}", var);
println!("A child thread borrowing `var2`: {:?}", var2);
})
});
}).unwrap();
}
I was just trying to figure out, if there is s.spawn() to spawn nested threads, is there a reason not to just create a new thread scope, like I seem to need to in this example?
That should be fine. Your thread that created the inner scope will be blocked waiting for completion of all its spawns, but that's the price for borrowing vec2.
Just tried it, that's because you are trying to pass a join guard outside of the second thread. Add a semicolon . Also the &mut var2 causes some lifetime errors because Rust of the closure desugaring not coercing unique references, so I removed that too
use crossbeam_utils::thread;
fn main() {
let var = vec![1, 2, 3];
let mut var2 = vec![3,2,1];
thread::scope(|s| {
s.spawn(|s| {
// no `let var2 = &mut var2;` here
s.spawn(|s| {
println!("A child thread borrowing `var`: {:?}", var);
println!("A child thread borrowing `var2`: {:?}", var2);
}); // semicolon here, to prevent the join guard from living too long
});
}).unwrap();
}
If you need to mutably access var2 after spawning the second thread, then yes, you will need a nested scope.
But in that example, var2 is instantiated outside of the initial scope. In my use-case var2 is instantiated inside the first scope, though it doesn't need to be mutable:
use crossbeam_utils::thread;
fn main() {
let var = vec![1, 2, 3];
thread::scope(|s| {
// var 2 is actually defined here
let var2 = vec![3,2,1];
s.spawn(|s| {
s.spawn(|_| {
println!("A child thread borrowing `var`: {:?}", var);
println!("A child thread borrowing `var2`: {:?}", var2);
});
});
}).unwrap();
}
I can't tell right off if that would work in my situation. Here's the actual code:
// Loop through cron jobs and run them if necessary
for (schedule, scripts) in &self.lucky_metadata.cron_jobs {
let schedule: cron::Schedule = handle_err!(schedule.parse(), call);
// If this job should be run
if let Some(date) = schedule.after(&last_cron_tick).next() {
if date < now {
// Run the job in its own thread
thread_scope(|s| {
s.spawn(|_| {
// For every script in the job
for script in scripts {
let hook_name = "cron";
// helper to run the script
macro_rules! run_script {
() => {
if let Err(e) = tools::run_charm_script(
&self,
hook_name,
&script,
environment,
) {
job_sender
.send(Err(e))
.expect("Channel dropped prematurely");
return Ok(());
}
// If docker is enabled, update container configuration
if self.lucky_metadata.use_docker {
if let Err(e) = tools::apply_container_updates(self) {
job_sender
.send(Err(e))
.expect("Channel dropped prematurely");
return Ok(());
}
}
};
}
// If the script is asynchronous
if script.is_async {
// Spawn it in another thread
thread_scope(|s| {
s.spawn(|_| {
run_script!();
Ok::<(), Void>(())
});
})
.expect("Panic in scoped thread");
// If the script is synchronous
} else {
// Run it in place
run_script!();
}
}
Ok::<(), Void>(())
});
})
.expect("Panic in scoped thread");
}
}
}
I'm iterating over cron schedules, spawning each cron job in its own thread, and potentially spawning asynchronous scripts for each cron job.
let environment = &environment;
let job_sender = &job_sender;
Right before the macro, and change the if to
// If the script is asynchronous
if script.is_async {
// Spawn it in another thread
s.spawn(move |_| {
// move closure, this is why I asked to rebind
// `environment` and `job_sender` so that
// don't get moved into the sub-threads
run_script!();
Ok::<(), Void>(())
});
.expect("Panic in scoped thread");
// If the script is synchronous
} else {
// Run it in place
run_script!();
}
Ah, OK, yeah. I did let job_sender = &job_sender and I happened to already be doing let environment = &environment. The final result ( I'll include the whole function this time for clarity ):
fn cron_tick(
&self,
call: &mut dyn rpc::Call_CronTick,
juju_context_id: String,
) -> varlink::Result<()> {
// Set the Juju context
std::env::set_var("JUJU_CONTEXT_ID", &juju_context_id);
log::trace!("Cron tick");
// Create environment map
let mut environment: HashMap<String, String> = HashMap::new();
environment.insert("JUJU_CONTEXT_ID".into(), juju_context_id);
// Make environment a reference ( so it can be used in threads )
let environment = &environment;
// Get the last cron tick time and the current time
let mut last_cron_tick = self.last_cron_tick.lock().unwrap();
let now = Local::now();
// Create a channel used to transefer our job results from their threads
let (job_sender, job_receiver) = unbounded_channel();
// Loop through cron jobs and run them if necessary
for (schedule, scripts) in &self.lucky_metadata.cron_jobs {
let schedule: cron::Schedule = handle_err!(schedule.parse(), call);
// If this job should be run
if let Some(date) = schedule.after(&last_cron_tick).next() {
if date < now {
// Run the job in its own thread
let job_sender = &job_sender;
thread_scope(|s| {
s.spawn(|ss| {
// For every script in the job
for script in scripts {
let hook_name = "cron";
// helper to run the script
macro_rules! run_script {
() => {
if let Err(e) = tools::run_charm_script(
&self,
hook_name,
&script,
environment,
) {
job_sender
.send(Err(e))
.expect("Channel dropped prematurely");
return Ok(());
}
// If docker is enabled, update container configuration
if self.lucky_metadata.use_docker {
if let Err(e) = tools::apply_container_updates(self) {
job_sender
.send(Err(e))
.expect("Channel dropped prematurely");
return Ok(());
}
}
};
}
// If the script is asynchronous
if script.is_async {
// Spawn it in another thread
ss.spawn(move |_| {
run_script!();
Ok::<(), Void>(())
});
// If the script is synchronous
} else {
// Run it in place
run_script!();
}
}
Ok::<(), Void>(())
});
})
.expect("Panic in scoped thread");
}
}
}
// Close the channel
drop(job_sender);
// Loop through job results
for job_result in job_receiver.iter() {
// Handle any errors
handle_err!(job_result, call);
}
// Update the last cron tick
*last_cron_tick = Local::now();
// Unset the Juju context as it will be invalid when the cron tick command exits
std::env::remove_var("JUJU_CONTEXT_ID");
// Reply empty
call.reply()
}
. Also the