1//! The sync worker's and background's loops: threads natively, and in the browser, which
2//! gives a module one thread, tasks on its event loop. Native waits block their thread, so
3//! `complete` runs the block to the end in one poll.
4
5use std::{io, time::Duration};
6
7#[cfg(not(target_arch = "wasm32"))]
8pub(crate) use std::{
9 sync::mpsc::{Receiver, SyncSender as Sender},
10 thread::JoinHandle,
11};
12
13#[cfg(not(target_arch = "wasm32"))]
14pub(crate) type Answer<T> = std::sync::mpsc::Sender<T>;
15
16#[cfg(not(target_arch = "wasm32"))]
17pub(crate) fn response<T>() -> (Answer<T>, std::sync::mpsc::Receiver<T>) {
18 std::sync::mpsc::channel()
19}
20
21/// Runs native blocking work; browser callers are refused before the future is polled.
22pub(crate) fn blocking<T>(work: impl Future<Output = T>) -> io::Result<T> {
23 #[cfg(not(target_arch = "wasm32"))]
24 return Ok(complete(work));
25 #[cfg(target_arch = "wasm32")]
26 {
27 drop(work);
28 Err(io::ErrorKind::WouldBlock.into())
29 }
30}
31
32#[cfg(all(feature = "live", not(target_arch = "wasm32")))]
33pub(crate) async fn answered<T>(
34 receiver: &std::sync::mpsc::Receiver<T>,
35 timeout: Duration,
36) -> Result<T, std::sync::mpsc::RecvTimeoutError> {
37 receiver.recv_timeout(timeout)
38}
39
40/// A wake that waits, at most one, as `mpsc::sync_channel(1)` keeps one.
41#[cfg(not(target_arch = "wasm32"))]
42pub(crate) fn channel() -> (Sender<()>, Receiver<()>) {
43 std::sync::mpsc::sync_channel(1)
44}
45
46/// Waits for a wake or `timeout`, without one for ever.
47#[cfg(not(target_arch = "wasm32"))]
48pub(crate) async fn wait(receiver: &Receiver<()>, timeout: Option<Duration>) {
49 let _ = match timeout {
50 Some(timeout) => receiver.recv_timeout(timeout).ok(),
51 None => receiver.recv().ok(),
52 };
53}
54
55/// Runs the loop `start` makes, whose waits block, on a thread of its own named `name`.
56#[cfg(not(target_arch = "wasm32"))]
57pub(crate) fn spawn<T: Send + 'static, F: Future<Output = T>>(
58 name: &str,
59 start: impl FnOnce() -> F + Send + 'static,
60) -> io::Result<JoinHandle<T>> {
61 std::thread::Builder::new()
62 .name(name.into())
63 .spawn(move || complete(start()))
64}
65
66/// `work`'s output, where nothing it awaits is pending.
67#[cfg(not(target_arch = "wasm32"))]
68pub(crate) fn complete<T>(work: impl Future<Output = T>) -> T {
69 let mut work = std::pin::pin!(work);
70 match work
71 .as_mut()
72 .poll(&mut std::task::Context::from_waker(std::task::Waker::noop()))
73 {
74 std::task::Poll::Ready(output) => output,
75 std::task::Poll::Pending => unreachable!("Native waits block rather than pend"),
76 }
77}
78
79/// Polls a synchronous entry point once; browser I/O must use the async entry point.
80pub(crate) fn ready<T>(work: impl Future<Output = T>) -> io::Result<T> {
81 let mut work = std::pin::pin!(work);
82 match work
83 .as_mut()
84 .poll(&mut std::task::Context::from_waker(std::task::Waker::noop()))
85 {
86 std::task::Poll::Ready(output) => Ok(output),
87 std::task::Poll::Pending => Err(io::ErrorKind::WouldBlock.into()),
88 }
89}
90
91#[cfg(target_arch = "wasm32")]
92pub(crate) use web::*;
93
94#[cfg(target_arch = "wasm32")]
95mod web {
96 use super::*;
97 use std::{
98 sync::{Arc, Mutex},
99 task::{Poll, Waker},
100 };
101 use wasm_bindgen::{JsCast, prelude::*};
102
103 struct Response<T> {
104 value: Option<T>,
105 closed: bool,
106 waiting: Option<Waker>,
107 }
108
109 pub(crate) struct Answer<T>(Arc<Mutex<Response<T>>>);
110 pub(crate) struct Answered<T>(Arc<Mutex<Response<T>>>);
111
112 pub(crate) fn response<T>() -> (Answer<T>, Answered<T>) {
113 let response = Arc::new(Mutex::new(Response {
114 value: None,
115 closed: false,
116 waiting: None,
117 }));
118 (Answer(response.clone()), Answered(response))
119 }
120
121 impl<T> Answer<T> {
122 pub(crate) fn send(self, value: T) -> Result<(), ()> {
123 let wake = {
124 let mut response = self.0.lock().map_err(|_| ())?;
125 response.value = Some(value);
126 response.waiting.take()
127 };
128 if let Some(wake) = wake {
129 wake.wake();
130 }
131 Ok(())
132 }
133 }
134
135 impl<T> Drop for Answer<T> {
136 fn drop(&mut self) {
137 let wake = {
138 let mut response = self.0.lock().unwrap();
139 response.closed = true;
140 response.waiting.take()
141 };
142 if let Some(wake) = wake {
143 wake.wake();
144 }
145 }
146 }
147
148 #[cfg(feature = "live")]
149 pub(crate) async fn answered<T>(
150 receiver: &Answered<T>,
151 timeout: Duration,
152 ) -> Result<T, std::sync::mpsc::RecvTimeoutError> {
153 let deadline = web_time::Instant::now() + timeout;
154 let mut answer = std::pin::pin!(receive(receiver));
155 std::future::poll_fn(|context| {
156 if let Poll::Ready(answer) = answer.as_mut().poll(context) {
157 return Poll::Ready(
158 answer.map_err(|_| std::sync::mpsc::RecvTimeoutError::Disconnected),
159 );
160 }
161 if web_time::Instant::now() >= deadline {
162 return Poll::Ready(Err(std::sync::mpsc::RecvTimeoutError::Timeout));
163 }
164 after(
165 deadline.saturating_duration_since(web_time::Instant::now()),
166 context.waker().clone(),
167 );
168 Poll::Pending
169 })
170 .await
171 }
172
173 pub(crate) async fn receive<T>(
174 receiver: &Answered<T>,
175 ) -> Result<T, std::sync::mpsc::RecvError> {
176 std::future::poll_fn(|context| {
177 let mut response = receiver.0.lock().unwrap();
178 if let Some(value) = response.value.take() {
179 return Poll::Ready(Ok(value));
180 }
181 if response.closed {
182 return Poll::Ready(Err(std::sync::mpsc::RecvError));
183 }
184 response.waiting = Some(context.waker().clone());
185 Poll::Pending
186 })
187 .await
188 }
189
190 #[derive(Default)]
191 struct Bell {
192 rung: bool,
193 waiting: Option<Waker>,
194 }
195
196 pub(crate) struct Sender<T>(Arc<Mutex<Bell>>, std::marker::PhantomData<T>);
197 pub(crate) struct Receiver<T>(Arc<Mutex<Bell>>, std::marker::PhantomData<T>);
198
199 pub(crate) fn channel() -> (Sender<()>, Receiver<()>) {
200 let bell = Arc::new(Mutex::new(Bell::default()));
201 (
202 Sender(Arc::clone(&bell), Default::default()),
203 Receiver(bell, Default::default()),
204 )
205 }
206
207 impl Sender<()> {
208 pub(crate) fn try_send(&self, (): ()) -> Result<(), ()> {
209 let waiting = {
210 let mut bell = self.0.lock().map_err(|_| ())?;
211 bell.rung = true;
212 bell.waiting.take()
213 };
214 if let Some(waiting) = waiting {
215 waiting.wake();
216 }
217 Ok(())
218 }
219 }
220
221 /// Wakes `waker` after `delay` through the global scope's `setTimeout`.
222 fn after(delay: Duration, waker: Waker) {
223 let global = js_sys::global();
224 let Ok(set_timeout) = js_sys::Reflect::get(&global, &"setTimeout".into())
225 .and_then(|function| function.dyn_into::<js_sys::Function>())
226 else {
227 return waker.wake();
228 };
229 let wake = Closure::once_into_js(move || waker.wake());
230 let _ = set_timeout.call2(&global, &wake, &(delay.as_secs_f64() * 1e3).ceil().into());
231 }
232
233 pub(crate) async fn wait(receiver: &Receiver<()>, timeout: Option<Duration>) {
234 let deadline = timeout.map(|timeout| web_time::Instant::now() + timeout);
235 std::future::poll_fn(|context| {
236 let Ok(mut bell) = receiver.0.lock() else {
237 return Poll::Ready(());
238 };
239 if std::mem::take(&mut bell.rung)
240 || deadline.is_some_and(|deadline| deadline <= web_time::Instant::now())
241 {
242 return Poll::Ready(());
243 }
244 bell.waiting = Some(context.waker().clone());
245 if let Some(deadline) = deadline {
246 after(
247 deadline.saturating_duration_since(web_time::Instant::now()),
248 context.waker().clone(),
249 );
250 }
251 Poll::Pending
252 })
253 .await
254 }
255
256 /// A task the browser's event loop runs, with a wakeable completion receipt.
257 pub(crate) struct JoinHandle<T>(Answered<T>);
258
259 impl<T> JoinHandle<T> {
260 /// What the task ended with, if it has ended.
261 pub(crate) fn finished(self) -> Option<T> {
262 self.0.0.lock().ok()?.value.take()
263 }
264
265 pub(crate) async fn join(self) -> Result<T, std::sync::mpsc::RecvError> {
266 receive(&self.0).await
267 }
268 }
269
270 pub(crate) fn spawn<T: 'static, F: Future<Output = T> + 'static>(
271 _: &str,
272 start: impl FnOnce() -> F,
273 ) -> io::Result<JoinHandle<T>> {
274 let (ended, output) = response();
275 let work = start();
276 wasm_bindgen_futures::spawn_local(async move {
277 let _ = ended.send(work.await);
278 });
279 Ok(JoinHandle(output))
280 }
281}