Merge pull request #27 from gents83/job-system-optimization-13438576981680291281

Refactor Job System for Efficiency and Parallelization

Author: gents83

.cargo/config.toml

@@ -40,8 +40,8 @@ wasm-opt = ['-Oz']
 # `brew install michaeleisel/zld/zld`
 [target.x86_64-apple-darwin]
 rustflags = ["-C", "link-arg=-fuse-ld=lld"]
-[target.aarch64-apple-darwin]
-rustflags = ["-C", "link-arg=-fuse-ld=/opt/homebrew/opt/llvm/bin/ld64.lld"]
+#[target.aarch64-apple-darwin]
+#rustflags = ["-C", "link-arg=-fuse-ld=/opt/homebrew/opt/llvm/bin/ld64.lld"]
 
 [target.wasm32-unknown-unknown]
 rustflags = [

.github/workflows/rust-android.yml

@@ -44,6 +44,11 @@ jobs:
           cd crates
           cargo ndk -t arm64-v8a clippy --release --no-deps -- -D warnings
 
+      - name: "Run cargo test workspace"
+        run: |
+          cd crates
+          cargo test --workspace
+
       - name: "Run cargo build release"
         run: |
           cd crates

.github/workflows/rust-ios.yml

@@ -41,6 +41,10 @@ jobs:
           export SDKROOT=$(xcrun -sdk iphoneos --show-sdk-path)
           cargo clippy --manifest-path ./crates/Cargo.toml --target aarch64-apple-ios --release --no-deps -- -D warnings
 
+      - name: "Run cargo test workspace"
+        run: |
+          cargo test --workspace --manifest-path ./crates/Cargo.toml
+
       - name: "Run cargo build release"
         run: |
           export SDKROOT=$(xcrun -sdk iphoneos --show-sdk-path)

.github/workflows/rust-unix.yml

@@ -78,6 +78,9 @@ jobs:
       - name: "Run cargo clippy check"
         run: cargo clippy --manifest-path ./crates/Cargo.toml --no-deps -- -D warnings
 
+      - name: "Run cargo test workspace"
+        run: cargo test --workspace --manifest-path ./crates/Cargo.toml
+
       - name: "Run cargo build workspace release"
         run: cargo build --manifest-path ./crates/Cargo.toml --release
 

.github/workflows/rust-web.yml

@@ -62,6 +62,9 @@ jobs:
           cargo update
           cd ..
 
+      - name: "Run cargo test workspace"
+        run: cargo test --workspace --manifest-path ./crates/Cargo.toml
+
       - name: "Run cargo build workspace release wasm32"
         run: cargo build --manifest-path ./crates/Cargo.toml --release --target wasm32-unknown-unknown
 

.github/workflows/rust-windows.yml

@@ -81,6 +81,9 @@ jobs:
       - name: "Run cargo clippy check"
         run: cargo clippy --manifest-path ./crates/Cargo.toml --no-deps -- -D warnings
 
+      - name: "Run cargo test workspace"
+        run: cargo test --workspace --manifest-path ./crates/Cargo.toml
+
       - name: "Run cargo build workspace release"
         run: cargo build --manifest-path ./crates/Cargo.toml --release
 

crates/Cargo.toml

@@ -71,7 +71,6 @@ pyo3 = { git = "https://github.com/PyO3/pyo3", features = [
     "abi3-py310",
     "generate-import-lib",
 ] }
-rand = { git = "https://github.com/rust-random/rand", features = ["std"] }
 serde_json = { git = "https://github.com/serde-rs/json" }
 flexbuffers = { git = "https://github.com/google/flatbuffers" }
 ttf-parser = { git = "https://github.com/RazrFalcon/ttf-parser" }
@@ -137,7 +136,6 @@ rspirv = { version = "0.11" }
 js-sys = { version = "0.3.77" }
 wasm-bindgen = { version = "0.2.100" }
 wasm-bindgen-futures = { version = "0.4.43" }
-wasm-bindgen-test = { version = "0.3" }
 web-sys = { version = "0.3.77", features = [
     "Request",
     "Response",

crates/core/Cargo.toml

@@ -25,4 +25,4 @@ inox_time = { path = "../time" }
 inox_uid = { path = "../uid" }
 
 [target.'cfg(target_arch = "wasm32")'.dependencies]
-wasm-bindgen = { workspace = true }
+wasm-bindgen = { workspace = true }

crates/core/src/schedule/job.rs

@@ -15,6 +15,7 @@ use crate::Worker;
 const NUM_WORKER_THREADS: usize = 0;
 #[cfg(not(target_arch = "wasm32"))]
 const NUM_WORKER_THREADS: usize = 5;
+const LOW_PRIORITY_THREAD_RATIO: f32 = 0.5;
 
 pub type JobId = Uid;
 pub const INDEPENDENT_JOB_ID: JobId = inox_uid::generate_static_uid_from_string("IndependentJob");
@@ -34,12 +35,6 @@ impl Job {
         F: FnOnce() + Send + Sync + 'static,
     {
         pending_jobs.fetch_add(1, Ordering::SeqCst);
-        /*
-        debug_log(
-            "Adding job {:?} - remaining {:?}",
-            name,
-            pending_jobs.load(Ordering::SeqCst)
-        );*/
         Self {
             func: Some(Box::new(func)),
             pending_jobs,
@@ -53,32 +48,18 @@ impl Job {
 
     pub fn execute(mut self) {
         inox_profiler::scoped_profile!("Job {}", self.name);
-        /*
-        debug_log(
-            "Starting {:?} - remaining {:?}",
-            self.name.as_str(),
-            self.pending_jobs.load(Ordering::SeqCst)
-        );
-        */
         let f = self.func.take().unwrap();
         (f)();
 
         self.pending_jobs.fetch_sub(1, Ordering::SeqCst);
         self.name.clear();
-        /*
-        debug_log(
-            "Ending {:?} - remaining {:?}",
-            self.name.as_str(),
-            self.pending_jobs.load(Ordering::SeqCst)
-        );
-        */
     }
 }
 
 pub type JobHandlerRw = Arc<RwLock<JobHandler>>;
 pub type JobReceiverRw = Arc<Mutex<Receiver<Job>>>;
 
-#[derive(Debug)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum JobPriority {
     High = 0,
     Medium = 1,
@@ -113,7 +94,7 @@ impl Default for PrioChannel {
 #[derive(Default)]
 pub struct JobHandler {
     channel: [PrioChannel; JobPriority::Count as usize],
-    pending_jobs: HashMap<JobId, Arc<AtomicUsize>>,
+    pending_jobs: RwLock<HashMap<JobId, Arc<AtomicUsize>>>,
     workers: HashMap<String, Worker>,
 }
 
@@ -124,11 +105,12 @@ impl JobHandler {
     #[inline]
     fn get_pending_jobs_count(&self, job_category: &JobId) -> usize {
         inox_profiler::scoped_profile!("JobHandler::get_pending_jobs_count");
-        if let Some(pending_jobs) = self.pending_jobs.get(job_category) {
-            pending_jobs.load(Ordering::SeqCst)
-        } else {
-            0
+        if let Ok(pending_jobs) = self.pending_jobs.read() {
+            if let Some(pending_jobs) = pending_jobs.get(job_category) {
+                return pending_jobs.load(Ordering::SeqCst);
+            }
         }
+        0
     }
     #[inline]
     fn get_job_with_priority(&self, job_priority: JobPriority) -> Option<Job> {
@@ -145,29 +127,43 @@ impl JobHandler {
         }
     }
 
-    fn add_worker(&mut self, name: &str, can_continue: &Arc<AtomicBool>) -> &mut Worker {
+    fn add_worker(
+        &mut self,
+        name: &str,
+        can_continue: &Arc<AtomicBool>,
+        receivers: Vec<JobReceiverRw>,
+    ) -> &mut Worker {
         let key = String::from(name);
         let w = self.workers.entry(key).or_default();
         if !w.is_started() {
-            w.start(
-                name,
-                can_continue,
-                self.channel
-                    .iter()
-                    .map(|channel| channel.receiver.clone())
-                    .collect(),
-            );
+            w.start(name, can_continue, receivers);
         }
         w
     }
 
     #[inline]
     fn setup_worker_threads(&mut self, can_continue: &Arc<AtomicBool>) {
-        #[allow(clippy::absurd_extreme_comparisons)]
         if NUM_WORKER_THREADS > 0 {
-            #[allow(clippy::reversed_empty_ranges)]
-            for i in 1..NUM_WORKER_THREADS + 1 {
-                self.add_worker(format!("Worker{i}").as_str(), can_continue);
+            // High priority jobs are mandatory and should be executed as fast as possible
+            // Low priority jobs are non-mandatory and should not block the frame
+            // We can set a ratio of threads that can execute Low priority jobs
+            let num_low_priority_workers = (NUM_WORKER_THREADS as f32 * LOW_PRIORITY_THREAD_RATIO).ceil() as usize;
+            let num_low_priority_workers = num_low_priority_workers.max(1);
+
+            for i in 0..NUM_WORKER_THREADS {
+                let mut receivers = vec![
+                    self.channel[JobPriority::High as usize].receiver.clone(),
+                    self.channel[JobPriority::Medium as usize].receiver.clone(),
+                ];
+                if i < num_low_priority_workers {
+                    receivers.push(self.channel[JobPriority::Low as usize].receiver.clone());
+                }
+
+                self.add_worker(
+                    format!("Worker{i}").as_str(),
+                    can_continue,
+                    receivers,
+                );
             }
         }
     }
@@ -177,10 +173,12 @@ impl JobHandler {
         for (_name, w) in self.workers.iter_mut() {
             w.stop();
         }
-        self.pending_jobs.iter().for_each(|(_, pending_jobs)| {
-            pending_jobs.store(0, Ordering::SeqCst);
-        });
-        self.pending_jobs.clear();
+        if let Ok(mut pending_jobs) = self.pending_jobs.write() {
+            pending_jobs.iter().for_each(|entry| {
+                entry.1.store(0, Ordering::SeqCst);
+            });
+            pending_jobs.clear();
+        }
 
         self.channel.iter_mut().for_each(|c| {
             while let Some(j) = c.receiver.get_job() {
@@ -190,7 +188,7 @@ impl JobHandler {
     }
 
     fn add_job<F>(
-        &mut self,
+        &self,
         job_category: &JobId,
         job_name: &str,
         job_priority: JobPriority,
@@ -199,13 +197,22 @@ impl JobHandler {
         F: FnOnce() + Send + Sync + 'static,
     {
         inox_profiler::scoped_profile!("JobHandler::add_job[{}]", job_name);
-        let pending_jobs = self
-            .pending_jobs
-            .entry(*job_category)
-            .or_insert_with(|| Arc::new(AtomicUsize::new(0)))
-            .clone();
+        let pending_jobs = {
+            let read_lock = self.pending_jobs.read().unwrap();
+            read_lock.get(job_category).cloned()
+        };
+        let pending_jobs = pending_jobs.unwrap_or_else(|| {
+            let mut write_lock = self.pending_jobs.write().unwrap();
+            write_lock
+                .entry(*job_category)
+                .or_insert_with(|| Arc::new(AtomicUsize::new(0)))
+                .clone()
+        });
+
         let job = Job::new(job_name, func, pending_jobs);
         self.channel[job_priority as usize].sender.send(job).ok();
+        // Wake up all workers as we don't know which one is sleeping on this priority
+        // Using unpark is cheap
         self.workers.iter().for_each(|(_n, w)| {
             w.wakeup();
         });
@@ -229,7 +236,7 @@ impl JobHandlerTrait for JobHandlerRw {
     where
         F: FnOnce() + Send + Sync + 'static,
     {
-        self.write()
+        self.read()
             .unwrap()
             .add_job(job_category, job_name, job_priority, func);
     }

crates/core/src/schedule/worker.rs

@@ -4,6 +4,7 @@ use std::{
         Arc,
     },
     thread::{self, JoinHandle},
+    time::Duration,
 };
 
 use crate::{JobReceiverRw, JobReceiverTrait};
@@ -30,24 +31,20 @@ impl Worker {
             let t = builder
                 .spawn(move || {
                     inox_profiler::register_profiler_thread!();
-                    let receivers_count = receivers.len() as i32;
-                    let mut i;
                     loop {
-                        i = 0i32;
-                        while i >= 0 && i < receivers_count {
-                            if let Some(job) = receivers[i as usize].get_job() {
+                        let mut executed = false;
+                        for r in &receivers {
+                            if let Some(job) = r.get_job() {
                                 job.execute();
-                                //force exit from loop
-                                i = -1;
-                            } else {
-                                i += 1;
+                                executed = true;
+                                break;
                             }
                         }
-                        if !can_continue.load(Ordering::SeqCst) {
-                            return false;
-                        }
-                        if i >= 0 {
-                            std::thread::park();
+                        if !executed {
+                            if !can_continue.load(Ordering::SeqCst) {
+                                return false;
+                            }
+                            thread::park_timeout(Duration::from_millis(10));
                         }
                     }
                 })