scirs2_linalg/parallel/work_stealing/
queue.rs

1//! Work queue implementation for the work-stealing scheduler
2//!
3//! This module contains the WorkQueue struct and its implementation,
4//! which manages work items for individual worker threads.
5
6use std::collections::VecDeque;
7use std::time::Duration;
8
9use super::core::WorkItem;
10
11/// Work queue for a single worker thread
12#[derive(Debug)]
13pub struct WorkQueue<T: Clone> {
14    /// Double-ended queue for work items
15    pub items: VecDeque<WorkItem<T>>,
16    /// Number of items processed by this worker
17    pub processed_count: usize,
18    /// Total execution time for this worker
19    pub total_time: Duration,
20    /// Average execution time per item
21    pub avg_time: Duration,
22}
23
24impl<T: Clone> Default for WorkQueue<T> {
25    fn default() -> Self {
26        Self {
27            items: VecDeque::new(),
28            processed_count: 0,
29            total_time: Duration::ZERO,
30            avg_time: Duration::ZERO,
31        }
32    }
33}
34
35impl<T: Clone> WorkQueue<T> {
36    /// Add work item to the front of the queue (for local work)
37    pub fn push_front(&mut self, item: WorkItem<T>) {
38        self.items.push_front(item);
39    }
40
41    /// Add work item to the back of the queue (for stolen work)
42    #[allow(dead_code)]
43    pub fn push_back(&mut self, item: WorkItem<T>) {
44        self.items.push_back(item);
45    }
46
47    /// Take work from the front (local work)
48    pub fn pop_front(&mut self) -> Option<WorkItem<T>> {
49        self.items.pop_front()
50    }
51
52    /// Steal work from the back (work stealing)
53    pub fn steal_back(&mut self) -> Option<WorkItem<T>> {
54        if self.items.len() > 1 {
55            self.items.pop_back()
56        } else {
57            None
58        }
59    }
60
61    /// Update timing statistics
62    pub fn update_timing(&mut self, executiontime: Duration) {
63        self.processed_count += 1;
64        self.total_time += executiontime;
65        self.avg_time = self.total_time / self.processed_count as u32;
66    }
67
68    /// Get the current load (estimated remaining work time)
69    pub fn estimated_load(&self) -> Duration {
70        let base_time = if self.avg_time.is_zero() {
71            Duration::from_millis(1) // Default estimate
72        } else {
73            self.avg_time
74        };
75
76        self.items
77            .iter()
78            .map(|item| item.estimated_time.unwrap_or(base_time))
79            .sum()
80    }
81}
scirs2_linalg/parallel/work_stealing/queue.rs

scirs2_linalg/parallel/work_stealing/
queue.rs
