emulsion 12.0.0

A fast and minimalistic image viewer
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113

use std::sync::{
	atomic::{AtomicBool, Ordering},
	Arc, Mutex,
};
use std::thread::JoinHandle;
use std::time::Duration;

enum ActionState<InpT, OutT> {
	Ready,
	InputGiven(InpT),
	Pending,
	OutputReady(OutT),
}

struct Shared<InpT, OutT> {
	run: AtomicBool,
	state: Mutex<ActionState<InpT, OutT>>,
}

// struct Processor<InpT, OutT> {
//     shared: Arc<Shared<InpT, OutT>>,
// }
// impl<InpT, OutT> Processor<InpT, OutT> {
//     fn run<F: FnMut(InpT)->OutT>(&mut self, action: F) {

//     }
// }

fn run_processor<InpT, OutT, F: FnMut(InpT) -> OutT>(
	shared: Arc<Shared<InpT, OutT>>,
	mut action: F,
) {
	while shared.run.load(Ordering::SeqCst) {
		// Sleep to prevent buisy waiting
		std::thread::sleep(Duration::from_millis(5));
		let input = {
			let mut state = shared.state.lock().unwrap();
			if let ActionState::InputGiven(_) = &*state {
				let mut input = ActionState::Pending;
				std::mem::swap(&mut input, &mut *state);
				match input {
					ActionState::InputGiven(inp) => Some(inp),
					_ => unreachable!(),
				}
			} else {
				None
			}
		}; // let go of the mutex lock
		if let Some(input) = input {
			let output = action(input);
			let mut state = shared.state.lock().unwrap();
			if let ActionState::Pending = &*state {
				*state = ActionState::OutputReady(output);
			}
		}
	}
}

pub struct ParallelAction<InpT, OutT> {
	shared: Arc<Shared<InpT, OutT>>,
	join_handle: Option<JoinHandle<()>>,
}

impl<InpT: Send + 'static, OutT: Send + 'static> ParallelAction<InpT, OutT> {
	pub fn new<F: 'static + Send + FnMut(InpT) -> OutT>(action: F) -> ParallelAction<InpT, OutT> {
		let shared =
			Arc::new(Shared { run: AtomicBool::new(true), state: Mutex::new(ActionState::Ready) });
		let handle = {
			// let mut processor = Processor {
			//     shared: shared.clone(),
			// };
			let shared = shared.clone();
			std::thread::spawn(move || {
				run_processor(shared, action);
			})
		};

		ParallelAction { join_handle: Some(handle), shared }
	}

	/// Returns the given input if the thread is currently occupied
	pub fn give_input(&self, input: InpT) {
		let mut state = self.shared.state.lock().unwrap();
		*state = ActionState::InputGiven(input);
	}

	pub fn try_get_output(&self) -> Option<OutT> {
		let mut state = self.shared.state.lock().unwrap();
		if let ActionState::OutputReady(_) = &*state {
			let mut output = ActionState::Ready;
			std::mem::swap(&mut output, &mut *state);
			match output {
				ActionState::OutputReady(o) => Some(o),
				_ => unreachable!(),
			}
		} else {
			None
		}
	}

	pub fn is_ready(&self) -> bool {
		let state = self.shared.state.lock().unwrap();
		matches!(&*state, ActionState::Ready)
	}
}
impl<InpT, OutT> Drop for ParallelAction<InpT, OutT> {
	fn drop(&mut self) {
		self.shared.run.store(false, Ordering::SeqCst);
		if let Some(handle) = self.join_handle.take() {
			handle.join().unwrap();
		}
	}
}