1use std::sync::atomic::{AtomicBool, Ordering};
2use std::sync::Arc;
3use std::thread::{self, JoinHandle};
4use std::time::Duration;
5
6use crossbeam_channel::RecvTimeoutError;
7use parking_lot::RwLock;
8use tokio::runtime::Runtime as TokioRuntime;
9
10use crate::bus::{Bus, BusSender};
11use crate::cmd::Cmd;
12use crate::effect::EffectId;
13use crate::env::Environment;
14use crate::program::{Program, ReducerProgram};
15use crate::reducer::Reducer;
16use crate::safe_reducer::safe_reduce_update;
17use crate::sub::Sub;
18
19use super::config::RuntimeConfig;
20use super::error::{build_tokio, RuntimeError};
21use super::interpreter::{interpret_effects_async, InterpreterState};
22use super::rw_store::{RwStore, RwStoreBackend};
23use super::store::{StoreHub, StoreWork, StoreWorkUnwindGuard};
24use super::subscription;
25
26pub struct RwRuntime<S, M> {
28 pub state: Arc<RwLock<S>>,
29 pub bus: Bus<M>,
30 pub env: Environment,
31 backend: RwStoreBackend<S, M>,
32 shutdown: Arc<AtomicBool>,
33 _thread: Option<JoinHandle<()>>,
34 _tokio: Arc<TokioRuntime>,
35 _subscriptions: Arc<subscription::SubscriptionHandles>,
36}
37
38impl<S, M> RwRuntime<S, M>
39where
40 S: Send + Sync + 'static,
41 M: Send + 'static,
42{
43 pub fn from_program(
44 program: Program<S, M>,
45 env: Environment,
46 config: RuntimeConfig,
47 ) -> Result<Self, RuntimeError> {
48 Self::bootstrap(
49 program.init,
50 program.update,
51 program.subscriptions,
52 env,
53 config,
54 )
55 }
56
57 pub fn from_reducer_program<R>(
58 program: ReducerProgram<R>,
59 env: Environment,
60 config: RuntimeConfig,
61 ) -> Result<Self, RuntimeError>
62 where
63 R: Reducer<State = S, Action = M> + Send + Sync + 'static,
64 {
65 let reducer = Arc::new(program.reducer);
66 let init = program.init;
67 let subscriptions = program.subscriptions;
68 Self::bootstrap(
69 init,
70 move |state, msg| reducer.reduce(state, msg),
71 subscriptions,
72 env,
73 config,
74 )
75 }
76
77 fn bootstrap(
78 init: fn() -> (S, Cmd<M>),
79 update: impl Fn(&mut S, M) -> Cmd<M> + Send + Sync + 'static,
80 subscriptions: fn(&S) -> Sub<M>,
81 env: Environment,
82 config: RuntimeConfig,
83 ) -> Result<Self, RuntimeError> {
84 let RuntimeConfig {
85 bus_capacity,
86 worker_threads,
87 thread_name,
88 } = config;
89 let (initial_state, init_cmd) = init();
90 let state = Arc::new(RwLock::new(initial_state));
91 let bus = Bus::new(bus_capacity);
92 let shutdown = Arc::new(AtomicBool::new(false));
93 let tokio = build_tokio(worker_threads, thread_name)?;
94 let interpreter = InterpreterState::new();
95 let sub_handles = Arc::new(subscription::SubscriptionHandles::new());
96 let hub = StoreHub::new(bus.sender(), interpreter);
97 let backend = RwStoreBackend::new(state.clone(), hub);
98 let tx = bus.sender();
99 let init_handles = tokio.block_on(interpret_effects_async(
100 init_cmd.into_effects(),
101 tx.clone(),
102 env.clone(),
103 tokio.handle().clone(),
104 backend.clone(),
105 ));
106 tokio.handle().spawn(async move {
107 for join in init_handles {
108 let _ = join.await;
109 }
110 });
111
112 let receiver = bus.receiver().clone();
113 let subs_fn = subscriptions;
114 let state_for_thread = state.clone();
115 let env_for_thread = env.clone();
116 let shutdown_for_thread = shutdown.clone();
117 let tokio_for_thread = tokio.clone();
118 let backend_for_thread = backend.clone();
119 let tx_for_thread = tx;
120 let subs_registry = sub_handles.clone();
121
122 let thread = thread::Builder::new()
123 .name(thread_name.to_string())
124 .spawn(move || {
125 let rt = tokio_for_thread;
126 let sync_subs = |state: &S| {
127 let sub = subs_fn(state);
128 subscription::sync_subscriptions(
129 &sub,
130 &subs_registry,
131 rt.handle().clone(),
132 tx_for_thread.clone(),
133 backend_for_thread.clone(),
134 shutdown_for_thread.clone(),
135 );
136 };
137
138 sync_subs(&state_for_thread.read());
139
140 while !shutdown_for_thread.load(Ordering::Relaxed) {
141 match receiver.recv_timeout(Duration::from_millis(50)) {
142 Ok(msg) => {
143 let mut unwind = StoreWorkUnwindGuard::new(&backend_for_thread);
144 let cmd = {
145 let mut guard = state_for_thread.write();
146 match safe_reduce_update(&mut *guard, |s, a| update(s, a), msg) {
147 Ok(cmd) => cmd,
148 Err(_) => {
149 drop(guard);
150 backend_for_thread.notify_state();
151 continue;
152 }
153 }
154 };
155 unwind.disarm();
156 backend_for_thread.notify_state();
157 let handles = rt.block_on(interpret_effects_async(
158 cmd.into_effects(),
159 tx_for_thread.clone(),
160 env_for_thread.clone(),
161 rt.handle().clone(),
162 backend_for_thread.clone(),
163 ));
164 if handles.is_empty() {
165 backend_for_thread.end_store_work();
166 } else {
167 let backend_wait = backend_for_thread.clone();
168 rt.handle().spawn(async move {
169 for join in handles {
170 let _ = join.await;
171 }
172 backend_wait.end_store_work();
173 });
174 }
175
176 sync_subs(&state_for_thread.read());
177 }
178 Err(RecvTimeoutError::Timeout) => continue,
179 Err(RecvTimeoutError::Disconnected) => break,
180 }
181 }
182 })
183 .map_err(RuntimeError::ThreadSpawn)?;
184
185 Ok(Self {
186 state,
187 bus,
188 env,
189 backend,
190 shutdown,
191 _thread: Some(thread),
192 _tokio: tokio,
193 _subscriptions: sub_handles,
194 })
195 }
196
197 pub fn rw_store(&self) -> RwStore<S, M>
198 where
199 S: Clone + Send + Sync + 'static,
200 {
201 self.backend.rw_store()
202 }
203
204 pub fn dispatch(&self, msg: M) {
205 let _ = self.bus.sender().send_blocking(msg);
206 }
207
208 pub fn sender(&self) -> BusSender<M> {
209 self.bus.sender()
210 }
211
212 pub fn cancel(&self, id: EffectId) {
213 if let Some(handle) = self.backend.hub.interpreter.cancel_tokens.lock().remove(&id) {
214 handle.abort();
215 }
216 }
217
218 pub fn shutdown(self) {
219 self.shutdown.store(true, Ordering::Relaxed);
220 self._subscriptions.abort_all();
221 if let Some(handle) = self._thread {
222 let _ = handle.join();
223 }
224 }
225}
226
227#[cfg(test)]
228mod tests {
229 use super::*;
230 use crate::panic_on_state_clone;
231
232 panic_on_state_clone! {
233 #[derive(Default)]
234 struct Counter {
235 n: i32,
236 }
237 }
238
239 fn init() -> (Counter, Cmd<i32>) {
240 (Counter::default(), Cmd::none())
241 }
242
243 fn update(s: &mut Counter, msg: i32) -> Cmd<i32> {
244 s.n += msg;
245 Cmd::none()
246 }
247
248 fn subs(_: &Counter) -> Sub<i32> {
249 Sub::none()
250 }
251
252 fn boot(program: Program<Counter, i32>) -> RwRuntime<Counter, i32> {
253 match RwRuntime::from_program(program, Environment::new(), RuntimeConfig::new(16)) {
254 Ok(runtime) => runtime,
255 Err(err) => panic!("test runtime bootstrap failed: {err}"),
256 }
257 }
258
259 #[test]
260 fn rw_runtime_applies_dispatched_messages() {
261 let runtime = boot(Program::new(init, update, subs));
262 runtime.dispatch(3);
263 runtime.dispatch(4);
264 std::thread::sleep(Duration::from_millis(200));
265 assert_eq!(runtime.state.read().n, 7);
266 runtime.shutdown();
267 }
268
269 #[test]
270 fn read_store_allows_concurrent_reads() {
271 let runtime = boot(Program::new(init, update, subs));
272 let store = runtime.rw_store();
273 store.dispatch(5);
274 std::thread::sleep(Duration::from_millis(100));
275 assert_eq!(store.read_store().with_read(|s| s.n), 5);
276 runtime.shutdown();
277 }
278}