what the fuck did I make

This commit is contained in:
Quaternions 2023-10-04 02:58:02 -07:00
parent 6fc585e01e
commit f320520292

View File

@ -1,59 +1,65 @@
use std::thread; use std::thread;
use std::sync::{mpsc,Arc}; use std::sync::{mpsc,Arc};
use parking_lot::{Mutex,Condvar}; use parking_lot::{Mutex,Condvar};
use std::sync::atomic::{AtomicBool, Ordering};
struct Worker { struct Worker<Task:Send> {
id: usize, sender: mpsc::Sender<Task>,
receiver: Arc<(Mutex<mpsc::Receiver<Task>>, Condvar)>, receiver: Arc<(Mutex<mpsc::Receiver<Task>>,Condvar)>,
is_active: Arc<AtomicBool>,
} }
impl Worker { impl<Task:Send+'static> Worker<Task> {
fn new(id: usize, receiver: Arc<(Mutex<mpsc::Receiver<Task>>, Condvar)>, is_active: Arc<AtomicBool>) -> Worker { fn new() -> Self {
Worker { id, receiver, is_active } let (sender, receiver) = mpsc::channel::<Task>();
Self {
sender,
receiver:Arc::new((Mutex::new(receiver),Condvar::new())),
}
} }
fn start(self) { fn send(&self,task:Task)->Result<(), mpsc::SendError<Task>>{
let ret=self.sender.send(task);
self.receiver.1.notify_one();
ret
}
fn start(&self) {
let receiver=self.receiver.clone();
thread::spawn(move || { thread::spawn(move || {
loop{
loop { loop {
match self.receiver.0.lock().recv() { match receiver.0.lock().recv() {
Ok(task) => { Ok(_task) => {
println!("Worker {} got a task: {}", self.id, task); println!("Worker got a task");
// Process the task // Process the task
} }
Err(_) => { Err(_) => {
println!("Worker {} stopping.", self.id); println!("Worker stopping.",);
break; break;
} }
} }
} }
receiver.1.wait(&mut receiver.0.lock());
// Set is_active to false when the worker is done }
self.is_active.store(false, Ordering::SeqCst);
self.receiver.1.notify_one();
}); });
} }
} }
type Task = String;
#[test] #[test]
fn test_worker() { fn test_worker() {
let (sender, receiver) = mpsc::channel::<Task>();
let receiver = Arc::new((Mutex::new(receiver), Condvar::new()));
let is_active = Arc::new(AtomicBool::new(true));
// Create the worker thread // Create the worker thread
let worker = Worker::new(1, Arc::clone(&receiver), Arc::clone(&is_active)); let worker = Worker::new();
// Start the worker thread // Start the worker thread
worker.start(); worker.start();
// Send tasks to the worker // Send tasks to the worker
for i in 0..5 { for i in 0..5 {
let task = format!("Task {}", i); let task = crate::instruction::TimedInstruction{
sender.send(task).unwrap(); time:0,
instruction:crate::body::PhysicsInstruction::StrafeTick,
};
worker.send(task).unwrap();
} }
// Optional: Signal the worker to stop (in a real-world scenario) // Optional: Signal the worker to stop (in a real-world scenario)
@ -62,17 +68,10 @@ fn test_worker() {
// Sleep to allow the worker thread to finish processing // Sleep to allow the worker thread to finish processing
thread::sleep(std::time::Duration::from_secs(2)); thread::sleep(std::time::Duration::from_secs(2));
// Check if the worker is still active
let is_worker_active = is_active.load(Ordering::SeqCst);
if !is_worker_active {
// If the worker is done, signal it to process a new task
is_active.store(true, Ordering::SeqCst);
thread::sleep(std::time::Duration::from_secs(2));
// Send a new task // Send a new task
sender.send("New Task".to_string()).unwrap(); let task = crate::instruction::TimedInstruction{
} else { time:0,
println!("Worker is still active. Skipping new task."); instruction:crate::body::PhysicsInstruction::StrafeTick,
} };
worker.send(task).unwrap();
} }