cranpose_services/
image_loading.rs

1//! Images loaded from the network, a file or bytes the app holds, decoded off
2//! the UI thread at the size they are drawn, as Compose apps load them with
3//! Coil's `AsyncImage`.
4//!
5//! [`rememberAsyncImage`] gives a composable the [`ImageLoadState`] of an
6//! [`ImageRequest`]: `Loading` until the image is decoded, then `Ready` or
7//! `Failed`. All images share one [`ImageLoader`]: a memory cache bounded by
8//! the bytes of its decoded pixels, a disk cache of downloaded files with a
9//! size limit, and one download for every request of one image. A request
10//! whose composables all left the composition stops its download.
11//!
12//! The network goes through `local_http_client()`, so an app enables the
13//! `http-native` (or `web-http`) feature as it does for its own requests.
14
15use std::{
16    future::Future,
17    path::PathBuf,
18    pin::Pin,
19    sync::{Arc, Mutex, PoisonError},
20};
21
22use cranpose_core::{
23    LaunchedEffect, State, collections::map::HashMap, mutableStateOf, rememberKeyed,
24};
25use cranpose_macros::composable;
26use cranpose_ui_graphics::ImageBitmap;
27use futures_util::future::{FutureExt, Shared, WeakShared};
28
29#[cfg(not(target_arch = "wasm32"))]
30use crate::async_io::Signal;
31use crate::http::HttpClientRef;
32
33/// The image to load, and the largest size in pixels it is drawn at.
34#[derive(Clone, Debug, PartialEq, Eq, Hash)]
35pub struct ImageRequest {
36    source: ImageSource,
37    max_width: u32,
38    max_height: u32,
39}
40
41#[derive(Clone, Debug)]
42enum ImageSource {
43    /// `http://`, `https://` or `file://`, or a file path.
44    Url(Arc<str>),
45    /// Encoded bytes the app read itself, from a picked file or a bundled
46    /// asset, named by `key` for the memory cache.
47    Bytes { key: Arc<str>, bytes: Arc<[u8]> },
48}
49
50impl ImageSource {
51    /// What names the image: its URL, or the key of its bytes. Comparing
52    /// and hashing the name keeps a lookup from reading the bytes.
53    fn name(&self) -> (bool, &str) {
54        match self {
55            Self::Url(url) => (true, url),
56            Self::Bytes { key, .. } => (false, key),
57        }
58    }
59}
60
61impl PartialEq for ImageSource {
62    fn eq(&self, other: &Self) -> bool {
63        self.name() == other.name()
64    }
65}
66
67impl Eq for ImageSource {}
68
69impl std::hash::Hash for ImageSource {
70    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
71        self.name().hash(state);
72    }
73}
74
75impl ImageRequest {
76    /// An image at `url`: `http://`, `https://`, `file://` or a file path.
77    pub fn new(url: impl Into<Arc<str>>) -> Self {
78        Self {
79            source: ImageSource::Url(url.into()),
80            max_width: u32::MAX,
81            max_height: u32::MAX,
82        }
83    }
84
85    /// An image in encoded `bytes` (PNG, JPEG, WebP or GIF) the app already
86    /// holds. `key` names it in the memory cache; one key is one image.
87    pub fn bytes(key: impl Into<Arc<str>>, bytes: impl Into<Arc<[u8]>>) -> Self {
88        Self {
89            source: ImageSource::Bytes {
90                key: key.into(),
91                bytes: bytes.into(),
92            },
93            max_width: u32::MAX,
94            max_height: u32::MAX,
95        }
96    }
97
98    /// Decodes the image no larger than `width` × `height` pixels, keeping
99    /// its aspect ratio: a thumbnail of a large photo holds a thumbnail's
100    /// memory. Pass the size it is drawn at, in device pixels.
101    pub fn max_size(mut self, width: u32, height: u32) -> Self {
102        self.max_width = width.max(1);
103        self.max_height = height.max(1);
104        self
105    }
106
107    /// [`max_size`](Self::max_size), unless the request already has one: a
108    /// composable that knows the size it draws at fills it in, and an app
109    /// that asked for a size keeps it.
110    pub fn or_max_size(self, width: u32, height: u32) -> Self {
111        if self.max_width == u32::MAX && self.max_height == u32::MAX {
112            self.max_size(width, height)
113        } else {
114            self
115        }
116    }
117}
118
119/// Where a request is.
120#[derive(Clone, Debug, PartialEq)]
121pub enum ImageLoadState {
122    /// The image is still being read and decoded.
123    Loading,
124    /// The image has loaded and is ready to draw.
125    Ready(ImageBitmap),
126    /// Loading failed for the stated reason.
127    Failed(ImageLoadError),
128}
129
130/// Why an image did not load.
131#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
132pub enum ImageLoadError {
133    /// The image bytes could not be read from their source. The text describes the cause.
134    #[error("could not read the image: {0}")]
135    Read(String),
136    /// The bytes were read but could not be decoded as an image. The text describes the cause.
137    #[error("could not decode the image: {0}")]
138    Decode(String),
139}
140
141/// How much the shared [`ImageLoader`] keeps.
142#[derive(Clone, Debug, PartialEq, Eq)]
143pub struct ImageCacheLimits {
144    /// The bytes of decoded pixels the memory cache holds.
145    pub memory_bytes: usize,
146    /// The bytes of downloaded files the disk cache holds.
147    pub disk_bytes: u64,
148}
149
150impl Default for ImageCacheLimits {
151    fn default() -> Self {
152        Self {
153            memory_bytes: 64 << 20,
154            disk_bytes: 128 << 20,
155        }
156    }
157}
158
159type LoadResult = Result<ImageBitmap, ImageLoadError>;
160
161#[cfg(not(target_arch = "wasm32"))]
162type LoadFuture = Pin<Box<dyn Future<Output = LoadResult> + Send>>;
163#[cfg(target_arch = "wasm32")]
164type LoadFuture = Pin<Box<dyn Future<Output = LoadResult>>>;
165
166/// What a loader shares with its loads: the browser's futures run on its one
167/// thread, so there they share through `Rc`.
168#[cfg(not(target_arch = "wasm32"))]
169type Handle<T> = Arc<T>;
170#[cfg(target_arch = "wasm32")]
171type Handle<T> = std::rc::Rc<T>;
172
173/// Loads, decodes and caches images. A clone is the same loader;
174/// [`image_loader`] is the one every [`rememberAsyncImage`] uses.
175#[derive(Clone)]
176pub struct ImageLoader {
177    inner: Handle<LoaderState>,
178}
179
180struct LoaderState {
181    memory: Mutex<MemoryCache>,
182    disk: Option<Arc<DiskCache>>,
183    in_flight: Mutex<HashMap<ImageRequest, WeakShared<LoadFuture>>>,
184}
185
186impl ImageLoader {
187    /// A loader that keeps downloaded files in `disk_dir`, or none on disk
188    /// when it is `None`.
189    pub fn new(limits: ImageCacheLimits, disk_dir: Option<PathBuf>) -> Self {
190        Self {
191            inner: Handle::new(LoaderState {
192                memory: Mutex::new(MemoryCache::new(limits.memory_bytes)),
193                disk: disk_dir.map(|dir| Arc::new(DiskCache::new(dir, limits.disk_bytes))),
194                in_flight: Mutex::new(HashMap::default()),
195            }),
196        }
197    }
198
199    /// The decoded image when the memory cache holds it.
200    pub fn cached(&self, request: &ImageRequest) -> Option<ImageBitmap> {
201        lock(&self.inner.memory).get(request)
202    }
203
204    /// Loads `request`, joining a load of the same request already under way.
205    /// The load stops when every future it returned is dropped.
206    pub fn load(
207        &self,
208        client: HttpClientRef,
209        request: &ImageRequest,
210    ) -> impl Future<Output = LoadResult> + use<> {
211        let shared = self.shared_load(client, request);
212        async move {
213            match shared {
214                Ok(bitmap) => Ok(bitmap),
215                Err(load) => load.await,
216            }
217        }
218    }
219
220    /// The cached image, or the load every caller of `request` shares.
221    fn shared_load(
222        &self,
223        client: HttpClientRef,
224        request: &ImageRequest,
225    ) -> Result<ImageBitmap, Shared<LoadFuture>> {
226        if let Some(bitmap) = self.cached(request) {
227            return Ok(bitmap);
228        }
229        let mut in_flight = lock(&self.inner.in_flight);
230        if let Some(load) = in_flight.get(request).and_then(WeakShared::upgrade) {
231            return Err(load);
232        }
233        let loader = self.clone();
234        let owned = request.clone();
235        let load: LoadFuture = Box::pin(async move {
236            let result = loader.fetch(client, &owned).await;
237            lock(&loader.inner.in_flight).remove(&owned);
238            if let Ok(bitmap) = &result {
239                lock(&loader.inner.memory).insert(owned, bitmap.clone());
240            }
241            result
242        });
243        let load = load.shared();
244        if let Some(weak) = load.downgrade() {
245            in_flight.insert(request.clone(), weak);
246        }
247        Err(load)
248    }
249
250    async fn fetch(&self, client: HttpClientRef, request: &ImageRequest) -> LoadResult {
251        let (width, height) = (request.max_width, request.max_height);
252        match &request.source {
253            ImageSource::Bytes { bytes, .. } => {
254                decode_off_thread(Encoded::Bytes(Arc::clone(bytes)), width, height).await
255            }
256            ImageSource::Url(url) if !is_remote(url) => {
257                decode_off_thread(Encoded::File(local_path(url)), width, height).await
258            }
259            ImageSource::Url(url) => {
260                if let Some(disk) = &self.inner.disk
261                    && let Some(cached) = decode_off_thread(
262                        Encoded::Cached(Arc::clone(disk), Arc::clone(url)),
263                        width,
264                        height,
265                    )
266                    .await
267                    .ok()
268                {
269                    return Ok(cached);
270                }
271                let bytes: Arc<[u8]> = client
272                    .get_bytes(url)
273                    .await
274                    .map_err(|error| ImageLoadError::Read(error.to_string()))?
275                    .into();
276                let encoded = match &self.inner.disk {
277                    Some(disk) => Encoded::Download {
278                        disk: Arc::clone(disk),
279                        url: Arc::clone(url),
280                        bytes,
281                    },
282                    None => Encoded::Bytes(bytes),
283                };
284                decode_off_thread(encoded, width, height).await
285            }
286        }
287    }
288}
289
290/// The loader every [`rememberAsyncImage`] shares, with the default limits
291/// and its disk cache in the application's cache folder.
292#[cfg(not(target_arch = "wasm32"))]
293pub fn image_loader() -> ImageLoader {
294    static LOADER: std::sync::OnceLock<ImageLoader> = std::sync::OnceLock::new();
295    LOADER.get_or_init(default_loader).clone()
296}
297
298/// The loader every [`rememberAsyncImage`] shares, with the default limits.
299/// The browser has no disk cache for it; its own HTTP cache stands in.
300#[cfg(target_arch = "wasm32")]
301pub fn image_loader() -> ImageLoader {
302    crate::thread_state::with(|state| {
303        state
304            .image_loader
305            .borrow_mut()
306            .get_or_insert_with(default_loader)
307            .clone()
308    })
309}
310
311fn default_loader() -> ImageLoader {
312    let disk_dir = if cfg!(target_arch = "wasm32") {
313        None
314    } else {
315        crate::host::application_directories()
316            .ok()
317            .map(|directories| directories.cache.join("images"))
318    };
319    ImageLoader::new(ImageCacheLimits::default(), disk_dir)
320}
321
322/// Loads `request` through [`image_loader`] while the caller stays in the
323/// composition. The caller reads the returned state, so it recomposes when
324/// the image is ready or failed. An image the memory cache holds is `Ready`
325/// in the first composition.
326///
327/// ```ignore
328/// match rememberAsyncImage(ImageRequest::new(url).max_size(240, 240)).value() {
329///     ImageLoadState::Ready(bitmap) => Image(BitmapPainter(bitmap), ..),
330///     ImageLoadState::Loading => Placeholder(),
331///     ImageLoadState::Failed(_) => BrokenImage(),
332/// }
333/// ```
334#[composable]
335pub fn rememberAsyncImage(request: ImageRequest) -> State<ImageLoadState> {
336    let loader = image_loader();
337    let cached = loader.cached(&request);
338    // Keyed: another request starts from its own cached image or `Loading`,
339    // never from the image of the request before it.
340    let state = rememberKeyed(request.clone(), move |_| {
341        mutableStateOf(cached.map_or(ImageLoadState::Loading, ImageLoadState::Ready))
342    });
343    let client = crate::http::local_http_client().current();
344    LaunchedEffect(request.clone(), async move |_| {
345        if let Some(bitmap) = loader.cached(&request) {
346            state.set(ImageLoadState::Ready(bitmap));
347            return;
348        }
349        state.set(ImageLoadState::Loading);
350        state.set(match loader.load(client, &request).await {
351            Ok(bitmap) => ImageLoadState::Ready(bitmap),
352            Err(error) => ImageLoadState::Failed(error),
353        });
354    });
355    state.as_state()
356}
357
358fn is_remote(url: &str) -> bool {
359    url.starts_with("http://") || url.starts_with("https://")
360}
361
362fn local_path(url: &str) -> PathBuf {
363    PathBuf::from(url.strip_prefix("file://").unwrap_or(url))
364}
365
366/// Where the encoded image comes from, read on the decode worker.
367enum Encoded {
368    Bytes(Arc<[u8]>),
369    File(PathBuf),
370    /// The disk cache's copy, if it holds one.
371    Cached(Arc<DiskCache>, Arc<str>),
372    /// A download, which the worker also writes to the disk cache.
373    Download {
374        disk: Arc<DiskCache>,
375        url: Arc<str>,
376        bytes: Arc<[u8]>,
377    },
378}
379
380impl Encoded {
381    fn read(self) -> Result<Arc<[u8]>, ImageLoadError> {
382        match self {
383            Self::Bytes(bytes) => Ok(bytes),
384            Self::File(path) => std::fs::read(&path)
385                .map(Into::into)
386                .map_err(|error| ImageLoadError::Read(format!("{}: {error}", path.display()))),
387            Self::Cached(disk, url) => disk
388                .read(&url)
389                .map(Into::into)
390                .ok_or_else(|| ImageLoadError::Read(format!("{url} is not on disk"))),
391            Self::Download { disk, url, bytes } => {
392                disk.write(&url, &bytes);
393                Ok(bytes)
394            }
395        }
396    }
397}
398
399fn decode(encoded: Encoded, max_width: u32, max_height: u32) -> LoadResult {
400    let bytes = encoded.read()?;
401    let image = image::load_from_memory(&bytes)
402        .map_err(|error| ImageLoadError::Decode(error.to_string()))?;
403    let image = if image.width() > max_width || image.height() > max_height {
404        image.resize(max_width, max_height, image::imageops::FilterType::Triangle)
405    } else {
406        image
407    };
408    let rgba = image.into_rgba8();
409    let (width, height) = rgba.dimensions();
410    ImageBitmap::from_rgba8(width, height, rgba.into_raw())
411        .map_err(|error| ImageLoadError::Decode(error.to_string()))
412}
413
414/// Decodes on a worker thread, so a large image never holds a frame. The
415/// browser has no threads to hand the work to.
416#[cfg(not(target_arch = "wasm32"))]
417async fn decode_off_thread(encoded: Encoded, max_width: u32, max_height: u32) -> LoadResult {
418    let signal = Signal::new();
419    let done = signal.clone();
420    // Held while a caller waits: a job whose callers all left is skipped.
421    let waiting = Arc::new(());
422    let waiter = Arc::downgrade(&waiting);
423    let job: DecodeJob = Box::new(move || {
424        if waiter.strong_count() == 0 {
425            done.close();
426        } else {
427            done.set(decode(encoded, max_width, max_height));
428        }
429    });
430    if decode_queue().send(job).is_err() {
431        return Err(ImageLoadError::Decode("the decode threads stopped".into()));
432    }
433    let result = signal.wait().await;
434    drop(waiting);
435    result.unwrap_or_else(|| {
436        Err(ImageLoadError::Decode(
437            "the decode thread dropped the job".into(),
438        ))
439    })
440}
441
442#[cfg(target_arch = "wasm32")]
443async fn decode_off_thread(encoded: Encoded, max_width: u32, max_height: u32) -> LoadResult {
444    decode(encoded, max_width, max_height)
445}
446
447#[cfg(not(target_arch = "wasm32"))]
448type DecodeJob = Box<dyn FnOnce() + Send>;
449
450/// The queue of the decode threads: as many as the machine has cores, up to
451/// four, so a screen of thumbnails decodes side by side.
452#[cfg(not(target_arch = "wasm32"))]
453fn decode_queue() -> &'static std::sync::mpsc::Sender<DecodeJob> {
454    static QUEUE: std::sync::OnceLock<std::sync::mpsc::Sender<DecodeJob>> =
455        std::sync::OnceLock::new();
456    QUEUE.get_or_init(|| {
457        let (sender, receiver) = std::sync::mpsc::channel::<DecodeJob>();
458        let receiver = Arc::new(Mutex::new(receiver));
459        let threads = std::thread::available_parallelism().map_or(1, |cores| cores.get().min(4));
460        for index in 0..threads {
461            let receiver = Arc::clone(&receiver);
462            let started = std::thread::Builder::new()
463                .name(format!("cranpose-image-{index}"))
464                .spawn(move || {
465                    while let Ok(job) = lock(&receiver).recv() {
466                        job();
467                    }
468                });
469            if let Err(error) = started {
470                log::warn!("image loading: could not start a decode thread: {error}");
471            }
472        }
473        sender
474    })
475}
476
477/// Decoded images by request, the least recently used leaving first once
478/// their pixels pass the limit.
479struct MemoryCache {
480    entries: HashMap<ImageRequest, (ImageBitmap, u64)>,
481    bytes: usize,
482    limit: usize,
483    clock: u64,
484}
485
486impl MemoryCache {
487    fn new(limit: usize) -> Self {
488        Self {
489            entries: HashMap::default(),
490            bytes: 0,
491            limit,
492            clock: 0,
493        }
494    }
495
496    fn get(&mut self, request: &ImageRequest) -> Option<ImageBitmap> {
497        self.clock += 1;
498        let clock = self.clock;
499        self.entries.get_mut(request).map(|(bitmap, used)| {
500            *used = clock;
501            bitmap.clone()
502        })
503    }
504
505    fn insert(&mut self, request: ImageRequest, bitmap: ImageBitmap) {
506        let size = pixel_bytes(&bitmap);
507        if size > self.limit {
508            return;
509        }
510        self.clock += 1;
511        if let Some((old, _)) = self.entries.insert(request, (bitmap, self.clock)) {
512            self.bytes -= pixel_bytes(&old);
513        }
514        self.bytes += size;
515        while self.bytes > self.limit {
516            let Some(oldest) = self
517                .entries
518                .iter()
519                .min_by_key(|(_, (_, used))| *used)
520                .map(|(request, _)| request.clone())
521            else {
522                break;
523            };
524            if let Some((evicted, _)) = self.entries.remove(&oldest) {
525                self.bytes -= pixel_bytes(&evicted);
526            }
527        }
528    }
529}
530
531fn pixel_bytes(bitmap: &ImageBitmap) -> usize {
532    bitmap.width() as usize * bitmap.height() as usize * 4
533}
534
535/// Downloaded files, one per URL, named by the URL's hash and starting with
536/// the URL itself, so two URLs with one hash never read as each other. Past
537/// the limit the least recently read files go.
538struct DiskCache {
539    dir: PathBuf,
540    limit: u64,
541    /// The bytes the folder holds: scanned at the first write, then added up.
542    size: Mutex<Option<u64>>,
543}
544
545impl DiskCache {
546    fn new(dir: PathBuf, limit: u64) -> Self {
547        Self {
548            dir,
549            limit,
550            size: Mutex::new(None),
551        }
552    }
553
554    fn path(&self, url: &str) -> PathBuf {
555        self.dir.join(format!("{:016x}", fnv1a(url.as_bytes())))
556    }
557
558    fn read(&self, url: &str) -> Option<Vec<u8>> {
559        let path = self.path(url);
560        let mut contents = std::fs::read(&path).ok()?;
561        let header = contents.iter().position(|&byte| byte == b'\n')?;
562        if &contents[..header] != url.as_bytes() {
563            return None;
564        }
565        // Reading counts as use for the eviction order.
566        if let Ok(file) = std::fs::File::options().append(true).open(&path) {
567            let _ = file.set_modified(std::time::SystemTime::now());
568        }
569        contents.drain(..=header);
570        Some(contents)
571    }
572
573    fn write(&self, url: &str, bytes: &[u8]) {
574        if std::fs::create_dir_all(&self.dir).is_err() {
575            return;
576        }
577        let path = self.path(url);
578        let partial = path.with_extension("partial");
579        let mut contents = Vec::with_capacity(url.len() + 1 + bytes.len());
580        contents.extend_from_slice(url.as_bytes());
581        contents.push(b'\n');
582        contents.extend_from_slice(bytes);
583        let written = contents.len() as u64;
584        if std::fs::write(&partial, contents).is_err() || std::fs::rename(&partial, &path).is_err()
585        {
586            return;
587        }
588        let mut size = lock(&self.size);
589        let total = match *size {
590            Some(total) => total + written,
591            None => self.files().iter().map(|(_, len, _)| len).sum(),
592        };
593        *size = Some(if total > self.limit {
594            self.trim()
595        } else {
596            total
597        });
598    }
599
600    fn files(&self) -> Vec<(std::time::SystemTime, u64, PathBuf)> {
601        std::fs::read_dir(&self.dir)
602            .into_iter()
603            .flatten()
604            .filter_map(Result::ok)
605            .filter_map(|entry| {
606                let metadata = entry.metadata().ok()?;
607                Some((metadata.modified().ok()?, metadata.len(), entry.path()))
608            })
609            .collect()
610    }
611
612    /// Removes the least recently used files until the rest fit the limit,
613    /// and returns the bytes left.
614    fn trim(&self) -> u64 {
615        let mut files = self.files();
616        let mut total: u64 = files.iter().map(|(_, len, _)| len).sum();
617        files.sort_by_key(|(modified, _, _)| *modified);
618        for (_, len, path) in files {
619            if total <= self.limit {
620                break;
621            }
622            if std::fs::remove_file(&path).is_ok() {
623                total -= len;
624            }
625        }
626        total
627    }
628}
629
630/// FNV-1a: a hash that stays the same across builds, which a file name must.
631fn fnv1a(bytes: &[u8]) -> u64 {
632    bytes.iter().fold(0xcbf2_9ce4_8422_2325, |hash, &byte| {
633        (hash ^ u64::from(byte)).wrapping_mul(0x0000_0100_0000_01b3)
634    })
635}
636
637fn lock<T>(mutex: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
638    mutex.lock().unwrap_or_else(PoisonError::into_inner)
639}
cranpose_services/image_loading.rs

cranpose_services/
image_loading.rs
