oxipng/src/evaluate.rs
Ingvar Stepanyan e38c759563
Switch to crossbeam-channel + rayon::spawn (#327)
* Switch to crossbeam-channel + rayon::spawn

* Remove thread_spawn for evaluation altogether

This allows to avoid a deadlock when there is only one Rayon thread, and doesn't sacrifice performance, since the caller of .get_result() had to always block on the iterator to be finished anyway, and all the messages are already sent from separate threads.

* Fix `verbose_mode` test

This one is easier to "fix", since we're in control of the rayon pool - just adding one extra thread to the default number to make sure we always can one root "spawn" call without stealing from the actual pool.
2020-11-11 12:01:13 -05:00

158 lines
5.4 KiB
Rust

//! Check if a reduction makes file smaller, and keep best reductions.
//! Works asynchronously when possible
use crate::atomicmin::AtomicMin;
use crate::deflate;
use crate::png::PngData;
use crate::png::PngImage;
use crate::png::STD_COMPRESSION;
use crate::png::STD_FILTERS;
use crate::png::STD_STRATEGY;
use crate::png::STD_WINDOW;
#[cfg(not(feature = "parallel"))]
use crate::rayon;
use crate::Deadline;
#[cfg(feature = "parallel")]
use crossbeam_channel::{unbounded, Receiver, Sender};
use rayon::prelude::*;
#[cfg(not(feature = "parallel"))]
use std::cell::RefCell;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering::SeqCst;
use std::sync::Arc;
struct Candidate {
image: PngData,
filter: u8,
// first wins tie-breaker
nth: usize,
}
impl Candidate {
fn cmp_key(&self) -> impl Ord {
(
self.image.idat_data.len(),
self.image.raw.data.len(),
self.image.raw.ihdr.bit_depth,
self.filter,
self.nth,
)
}
}
/// Collect image versions and pick one that compresses best
pub(crate) struct Evaluator {
deadline: Arc<Deadline>,
nth: AtomicUsize,
best_candidate_size: Arc<AtomicMin>,
/// images are sent to the caller thread for evaluation
#[cfg(feature = "parallel")]
eval_channel: (Sender<Candidate>, Receiver<Candidate>),
// in non-parallel mode, images are evaluated synchronously
#[cfg(not(feature = "parallel"))]
eval_best_candidate: RefCell<Option<Candidate>>,
}
impl Evaluator {
pub fn new(deadline: Arc<Deadline>) -> Self {
#[cfg(feature = "parallel")]
let eval_channel = unbounded();
Self {
deadline,
best_candidate_size: Arc::new(AtomicMin::new(None)),
nth: AtomicUsize::new(0),
#[cfg(feature = "parallel")]
eval_channel,
#[cfg(not(feature = "parallel"))]
eval_best_candidate: RefCell::new(None),
}
}
/// Wait for all evaluations to finish and return smallest reduction
/// Or `None` if all reductions were worse than baseline.
#[cfg(feature = "parallel")]
fn get_best_candidate(self) -> Option<Candidate> {
let (eval_send, eval_recv) = self.eval_channel;
drop(eval_send); // disconnect the sender, breaking the loop in the thread
eval_recv.into_iter().min_by_key(Candidate::cmp_key)
}
#[cfg(not(feature = "parallel"))]
fn get_best_candidate(self) -> Option<Candidate> {
self.eval_best_candidate.into_inner()
}
pub fn get_result(self) -> Option<PngData> {
self.get_best_candidate().map(|candidate| candidate.image)
}
/// Set baseline image. It will be used only to measure minimum compression level required
pub fn set_baseline(&self, image: Arc<PngImage>) {
self.try_image_inner(image, false)
}
/// Check if the image is smaller than others
pub fn try_image(&self, image: Arc<PngImage>) {
self.try_image_inner(image, true)
}
fn try_image_inner(&self, image: Arc<PngImage>, is_reduction: bool) {
let nth = self.nth.fetch_add(1, SeqCst);
// These clones are only cheap refcounts
let deadline = self.deadline.clone();
let best_candidate_size = self.best_candidate_size.clone();
// sends it off asynchronously for compression,
// but results will be collected via the message queue
#[cfg(feature = "parallel")]
let eval_send = self.eval_channel.0.clone();
rayon::spawn(move || {
let filters_iter = STD_FILTERS.par_iter().with_max_len(1);
// Updating of best result inside the parallel loop would require locks,
// which are dangerous to do in side Rayon's loop.
// Instead, only update (atomic) best size in real time,
// and the best result later without need for locks.
filters_iter.for_each(|&filter| {
if deadline.passed() {
return;
}
if let Ok(idat_data) = deflate::deflate(
&image.filter_image(filter),
STD_COMPRESSION,
STD_STRATEGY,
STD_WINDOW,
&best_candidate_size,
&deadline,
) {
best_candidate_size.set_min(idat_data.len());
// ignore baseline images after this point
if !is_reduction {
return;
}
// the rest is shipped to the evavluation/collection thread
let new = Candidate {
image: PngData {
idat_data,
raw: Arc::clone(&image),
},
filter,
nth,
};
#[cfg(feature = "parallel")]
{
eval_send.send(new).expect("send");
}
#[cfg(not(feature = "parallel"))]
{
match &mut *self.eval_best_candidate.borrow_mut() {
Some(prev) if prev.cmp_key() < new.cmp_key() => {}
best => *best = Some(new),
}
}
}
});
});
}
}