danvics/oxipng
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Switch to crossbeam-channel + rayon::spawn

Browse source
This commit is contained in:
Ingvar Stepanyan 2020-05-01 12:07:17 +01:00 committed by Josh Holmer
parent c612166c73
commit 567a3c6b1d
4 changed files with 59 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

38
Cargo.lock generated
View file

@ -162,6 +162,16 @@ dependencies = [
"cfg-if 1.0.0",
]
[[package]]
name = "crossbeam-channel"
version = "0.4.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b153fe7cbef478c567df0f972e02e6d736db11affe43dfc9c56a9374d1adfb87"
dependencies = [
"crossbeam-utils 0.7.2",
"maybe-uninit",
]
[[package]]
name = "crossbeam-channel"
version = "0.5.0"
@ -169,7 +179,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dca26ee1f8d361640700bde38b2c37d8c22b3ce2d360e1fc1c74ea4b0aa7d775"
dependencies = [
"cfg-if 1.0.0",
"crossbeam-utils",
"crossbeam-utils 0.8.0",
]
[[package]]
@ -180,7 +190,7 @@ checksum = "94af6efb46fef72616855b036a624cf27ba656ffc9be1b9a3c931cfc7749a9a9"
dependencies = [
"cfg-if 1.0.0",
"crossbeam-epoch",
"crossbeam-utils",
"crossbeam-utils 0.8.0",
]
[[package]]
@ -191,12 +201,23 @@ checksum = "ec0f606a85340376eef0d6d8fec399e6d4a544d648386c6645eb6d0653b27d9f"
dependencies = [
"cfg-if 1.0.0",
"const_fn",
"crossbeam-utils",
"crossbeam-utils 0.8.0",
"lazy_static",
"memoffset",
"scopeguard",
]
[[package]]
name = "crossbeam-utils"
version = "0.7.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c3c7c73a2d1e9fc0886a08b93e98eb643461230d5f1925e4036204d5f2e261a8"
dependencies = [
"autocfg",
"cfg-if 0.1.10",
"lazy_static",
]
[[package]]
name = "crossbeam-utils"
version = "0.8.0"
@ -320,6 +341,12 @@ dependencies = [
"cfg-if 0.1.10",
]
[[package]]
name = "maybe-uninit"
version = "2.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "60302e4db3a61da70c0cb7991976248362f30319e88850c487b9b95bbf059e00"
[[package]]
name = "memoffset"
version = "0.5.6"
@ -408,6 +435,7 @@ dependencies = [
"clap",
"cloudflare-zlib",
"crc",
"crossbeam-channel 0.4.4",
"image",
"indexmap",
"itertools",
@ -461,9 +489,9 @@ version = "1.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9ab346ac5921dc62ffa9f89b7a773907511cdfa5490c572ae9be1be33e8afa4a"
dependencies = [
"crossbeam-channel",
"crossbeam-channel 0.5.0",
"crossbeam-deque",
"crossbeam-utils",
"crossbeam-utils 0.8.0",
"lazy_static",
"num_cpus",
]

1
Cargo.toml
View file

@ -40,6 +40,7 @@ indexmap = "1.6.0"
libdeflater = { version = "0.5.0", optional = true }
log = "0.4.11"
stderrlog = { version = "0.5.0", optional = true }
crossbeam-channel = "0.4.4"
[dependencies.rayon]
optional = true

31
src/evaluate.rs
View file

@ -12,16 +12,14 @@ use crate::png::STD_WINDOW;
#[cfg(not(feature = "parallel"))]
use crate::rayon;
use crate::Deadline;
#[cfg(feature = "parallel")]
use crossbeam_channel::{bounded, 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;
#[cfg(feature = "parallel")]
use std::sync::mpsc::*;
use std::sync::Arc;
#[cfg(feature = "parallel")]
use std::thread;
struct Candidate {
image: PngData,
@ -49,19 +47,34 @@ pub(crate) struct Evaluator {
best_candidate_size: Arc<AtomicMin>,
/// images are sent to the thread for evaluation
#[cfg(feature = "parallel")]
eval_send: SyncSender<Candidate>,
eval_send: Sender<Candidate>,
// the thread helps evaluate images asynchronously
#[cfg(feature = "parallel")]
eval_thread: thread::JoinHandle<Option<Candidate>>,
eval_thread: Receiver<Option<Candidate>>,
// in non-parallel mode, images are evaluated synchronously
#[cfg(not(feature = "parallel"))]
eval_best_candidate: RefCell<Option<Candidate>>,
}
// Like `std::thread::spawn`, but uses Rayon to create a thread.
//
// This allows this function to work on targets like WebAssembly,
// where `std::thread::spawn` doesn't work but Rayon can thanks to its
// support for custom spawn handlers.
//
// Unlike `std::thread::spawn`, `rayon::spawn` doesn't support closure results
// (it doesn't return a `JoinHandle<T>`), so we simulate it with a crossbeam
// channel instead (where `Receiver` acts as a `JoinHandle`).
fn thread_spawn<T: Send + 'static>(f: impl FnOnce() -> T + Send + 'static) -> Receiver<T> {
let (tx, rx) = bounded(0);
rayon::spawn(move || tx.send(f()).unwrap());
rx
}
impl Evaluator {
pub fn new(deadline: Arc<Deadline>) -> Self {
#[cfg(feature = "parallel")]
let (tx, rx) = sync_channel(4);
let (tx, rx) = unbounded();
Self {
deadline,
best_candidate_size: Arc::new(AtomicMin::new(None)),
@ -69,7 +82,7 @@ impl Evaluator {
#[cfg(feature = "parallel")]
eval_send: tx,
#[cfg(feature = "parallel")]
eval_thread: thread::spawn(move || rx.into_iter().min_by_key(Candidate::cmp_key)),
eval_thread: thread_spawn(move || rx.into_iter().min_by_key(Candidate::cmp_key)),
#[cfg(not(feature = "parallel"))]
eval_best_candidate: RefCell::new(None),
}
@ -80,7 +93,7 @@ impl Evaluator {
#[cfg(feature = "parallel")]
fn get_best_candidate(self) -> Option<Candidate> {
drop(self.eval_send); // disconnect the sender, breaking the loop in the thread
self.eval_thread.join().expect("eval thread")
self.eval_thread.recv().expect("eval thread")
}
#[cfg(not(feature = "parallel"))]

6
tests/flags.rs
View file

@ -55,9 +55,9 @@ fn test_it_converts(
#[test]
fn verbose_mode() {
use crossbeam_channel::{unbounded, Sender};
use log::{set_logger, set_max_level, Level, LevelFilter, Log, Metadata, Record};
use std::cell::RefCell;
use std::sync::mpsc::{sync_channel, SyncSender};
// Rust runs tests in parallel by default.
// We want to make sure that we verify only logs from our test.
@ -66,7 +66,7 @@ fn verbose_mode() {
// initialise it with Some(sender) only on threads spawned within
// our test.
thread_local! {
static VERBOSE_LOGS: RefCell<Option<SyncSender<String>>> = RefCell::new(None);
static VERBOSE_LOGS: RefCell<Option<Sender<String>>> = RefCell::new(None);
}
struct LogTester;
@ -97,7 +97,7 @@ fn verbose_mode() {
let input = PathBuf::from("tests/files/verbose_mode.png");
let (output, opts) = get_opts(&input);
let (sender, receiver) = sync_channel(4);
let (sender, receiver) = unbounded();
let thread_init = move || {
// Initialise logs storage for all threads within our test.