rust/neon
1
0
Fork 0
mirror of https://github.com/neondatabase/neon.git synced 2026-05-18 05:30:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Cleanup

Browse Source
This commit is contained in:
Bojan Serafimov
2022-05-11 18:55:58 -04:00
parent 30376d87ac
commit 491a1870ce
1 changed files with 8 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
pageserver/src/jobs/scheduler.rs
View File

@@ -36,33 +36,26 @@ impl<J: Job> Sched<J> {
let shutdown_watcher = shutdown_watcher();
tokio::select! {
_ = shutdown_watcher => break,
w = self.worker.1.recv() => {
worker = self.worker.1.recv() => {
// Assign to next job in queue, if nonempty
println!("got worker");
if let Some(j) = jobs.pop_front() {
println!("found job from queue");
w.unwrap().0.send(j).await.unwrap();
worker.unwrap().0.send(j).await.unwrap();
} else {
println!("no jobs in queue");
workers.push(w.unwrap());
workers.push(worker.unwrap());
}
},
j = self.work.1.recv() => {
job = self.work.1.recv() => {
// Assign to first worker in pool, if nonempty
println!("got job");
if let Some(w) = workers.pop() {
println!("found worker in pool");
w.0.send(j.unwrap()).await.unwrap();
w.0.send(job.unwrap()).await.unwrap();
} else {
println!("no workers in pool");
jobs.push_back(j.unwrap());
jobs.push_back(job.unwrap());
}
},
r = self.report.1.recv() => {
report = self.report.1.recv() => {
// Reschedule job to run again
let send_work = self.work.0.clone();
let job = r.unwrap().for_job;
println!("rescheduling");
let job = report.unwrap().for_job;
tokio::spawn(async move {
sleep(Duration::from_millis(10)).await;
send_work.send(job).await.unwrap();