-
Notifications
You must be signed in to change notification settings - Fork 483
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[WIP] Create Scan parallel iterator #1036
Draft
arieluy
wants to merge
2
commits into
rayon-rs:main
Choose a base branch
from
arieluy:scan
base: main
Could not load branches
Branch not found: {{ refName }}
Could not load tags
Nothing to show
Are you sure you want to change the base?
Some commits from the old base branch may be removed from the timeline,
and old review comments may become outdated.
Draft
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit
Hold shift + click to select a range
File filter
Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,124 @@ | ||
use ndarray::{Array, Dim}; | ||
use rayon::iter::*; | ||
use std::time::{Duration, Instant}; | ||
use std::num::Wrapping; | ||
|
||
const SIZE: usize = 10000; | ||
|
||
enum Procs { | ||
Sequential, | ||
Parallel, | ||
} | ||
|
||
fn scan_sequential<T, P, I>(init: I, id: T, scan_op: P) -> Vec<T> | ||
where | ||
T: Clone, | ||
I: Fn() -> T, | ||
P: FnMut(&mut T, &T) -> Option<T>, | ||
{ | ||
let v = vec![init(); SIZE]; | ||
let scan = v.iter().scan(id, scan_op); | ||
scan.collect() | ||
} | ||
|
||
fn scan_parallel<T, P, I>(init: I, id: T, scan_op: P) -> Vec<T> | ||
where | ||
T: Clone + Send + Sync, | ||
I: Fn() -> T, | ||
P: Fn(&T, &T) -> T + Sync, | ||
{ | ||
let v = vec![init(); SIZE]; | ||
let scan = v.into_par_iter().with_min_len(SIZE / 100).scan(&scan_op, id); | ||
scan.collect() | ||
} | ||
|
||
/******* Addition with artificial delay *******/ | ||
|
||
const DELAY: Duration = Duration::from_nanos(10); | ||
fn wait() -> i32 { | ||
let time = Instant::now(); | ||
|
||
let mut sum = 0; | ||
while time.elapsed() < DELAY { | ||
sum += 1; | ||
} | ||
sum | ||
} | ||
|
||
fn scan_add(procs: Procs) -> Vec<i32> { | ||
let init = || 2; | ||
let id = 0; | ||
|
||
match procs { | ||
Procs::Sequential => { | ||
let f = |state: &mut i32, x: &i32| { | ||
test::black_box(wait()); | ||
*state += x; | ||
Some(*state) | ||
}; | ||
scan_sequential(init, id, f) | ||
} | ||
Procs::Parallel => { | ||
let f = |x: &i32, y: &i32| { | ||
test::black_box(wait()); | ||
*x + *y | ||
}; | ||
scan_parallel(init, id, f) | ||
} | ||
} | ||
} | ||
|
||
#[bench] | ||
fn scan_add_sequential(b: &mut test::Bencher) { | ||
b.iter(|| scan_add(Procs::Sequential)); | ||
} | ||
|
||
#[bench] | ||
fn scan_add_parallel(b: &mut test::Bencher) { | ||
b.iter(|| scan_add(Procs::Parallel)); | ||
} | ||
|
||
#[test] | ||
fn test_scan_add() { | ||
assert_eq!(scan_add(Procs::Sequential), scan_add(Procs::Parallel)); | ||
} | ||
|
||
/******** Matrix multiplication with wrapping arithmetic *******/ | ||
|
||
type Matrix = Array<Wrapping<i32>, Dim<[usize; 2]>>; | ||
fn scan_matmul(procs: Procs) -> Vec<Matrix> { | ||
const MAT_SIZE: usize = 50; | ||
let init = || { | ||
Array::from_iter((0..((MAT_SIZE * MAT_SIZE) as i32)).map(|x| Wrapping(x))) | ||
.into_shape((MAT_SIZE, MAT_SIZE)) | ||
.unwrap() | ||
}; | ||
let id = Array::eye(MAT_SIZE); | ||
|
||
match procs { | ||
Procs::Sequential => { | ||
let f = |state: &mut Matrix, x: &Matrix| { | ||
*state = state.dot(x); | ||
Some(state.clone()) | ||
}; | ||
|
||
scan_sequential(init, id, f) | ||
} | ||
Procs::Parallel => scan_parallel(init, id, |x, y| x.dot(y)), | ||
} | ||
} | ||
|
||
#[bench] | ||
fn scan_matmul_sequential(b: &mut test::Bencher) { | ||
b.iter(|| scan_matmul(Procs::Sequential)); | ||
} | ||
|
||
#[bench] | ||
fn scan_matmul_parallel(b: &mut test::Bencher) { | ||
b.iter(|| scan_matmul(Procs::Parallel)); | ||
} | ||
|
||
#[test] | ||
fn test_scan_matmul() { | ||
assert_eq!(scan_matmul(Procs::Sequential), scan_matmul(Procs::Parallel)); | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Let's start here, documenting at the API level, especially for a user who is looking for an equivalent to
Iterator::scan
. How would you describe what this does, and importantly how is it different than the sequential version? The signature ofF
is quite different, looking more like some flavor ofreduce
.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Good question, I wasn't exactly sure what the signature should be, and it could change. But it's similar to the way
reduce
compares tofold
. This should have an identical result to sequential scan when the operation doesn't have internal state, and is associative. If not, it won't make sense to run it in parallel.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Hi @cuviper, could you take another look at this?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I don't feel you addressed my first comment? The signature is part of it, but please make an attempt at adding documentation describing what this does. And frankly, I don't think many people even use
Iterator::scan
, so it's not sufficient to just reference that. Suppose this existed independently - describe what it does, and what points are important for the user to think about.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Sorry, I misunderstood you last time. I've added documentation.
I was hoping to get some feedback on the signature and general approach. These are the differences from the other ParallelIterator functions, and my rationale for them:
Fn(&Item, &Item) -> Item
, since when we call it iteratively, we need to keep each intermediate result and not consume it.scan_op
takes in references, we don't need multiple copies ofidentity
. So, I haveidentity
as typeItem
, rather thanFn() -> Item
, since it's simpler. The other functions useFn() -> Item
, though, and it would be easy to change.Do you have any preferences on those?