emulsion 8.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
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128

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();
			let has_input = match &*state {
				ActionState::InputGiven(_) => true,
				_ => false,
			};
			if has_input {
				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();
			let got_new_request = match &*state {
				ActionState::Pending => false,
				_ => true,
			};
			if !got_new_request {
				*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();
		let has_output = match &*state {
			ActionState::OutputReady(_) => true,
			_ => false,
		};
		if has_output {
			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();
		match &*state {
			ActionState::Ready => true,
			_ => false,
		}
	}
}
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();
		}
	}
}