History

unify trait bounds for parallel code: prefer Clone over Sync (#259)

Sebastian Thiel committed 4 years ago

Commit c805d0b1 parent 3c790e06 files · +24 −24Browse files

Unified · Split

6 files
  1. DEVELOPMENT.md +6 −0
  2. git-features/src/parallel/in_parallel.rs +4 −4
  3. git-features/src/parallel/mod.rs +2 −2
  4. git-features/src/parallel/reduce.rs +7 −13
  5. git-pack/src/data/output/count/objects/mod.rs +2 −2
  6. git-pack/src/data/output/entry/iter_from_counts.rs +3 −3
DEVELOPMENT.md+6−0
@@ -683,3 +683,9 @@ Please note that these are based on the following value system: · 682 unmodified lines
683 a single-threaded version).683 a single-threaded version).
684 - Maybe… after some benchmarks, we can entirely drop the single-threaded version if it's not significantly faster on a single thread (without thread primiives) than the684 - Maybe… after some benchmarks, we can entirely drop the single-threaded version if it's not significantly faster on a single thread (without thread primiives) than the
685 same with multi-threading (but a single-thread).685 same with multi-threading (but a single-thread).
686 - Servers would be split into readers and writers, where…
687 - …readers share a common pool and thus all maps, with lazy loading and refresh (but their pack-ids change due to that, and packs might disappear)
688 - …writers when…
689 - …cloning use an eagerly loaded Repository just for that particular clone for stable pack ids
690 - …fetches use a lazy-loaded Repository with refresh disabled, and full retries if the pack they were referring to goes away. Maybe there can be a policy for that to keep
691 pack ids stable despite refresh, which would also solve clones which could then lazy-load.
git-features/src/parallel/in_parallel.rs+4−4
@@ -22,8 +22,8 @@ pub fn join<O1: Send, O2: Send>(left: impl FnOnce() -> O1 + Send, right: impl Fn · 21 unmodified lines
22pub fn in_parallel<I, S, O, R>(22pub fn in_parallel<I, S, O, R>(
23 input: impl Iterator<Item = I> + Send,23 input: impl Iterator<Item = I> + Send,
24 thread_limit: Option<usize>,24 thread_limit: Option<usize>,
25 new_thread_state: impl Fn(usize) -> S + Send + Sync,25 new_thread_state: impl Fn(usize) -> S + Send + Clone,
26 consume: impl Fn(I, &mut S) -> O + Send + Sync,26 consume: impl Fn(I, &mut S) -> O + Send + Clone,
27 mut reducer: R,27 mut reducer: R,
28) -> Result<<R as Reduce>::Output, <R as Reduce>::Error>28) -> Result<<R as Reduce>::Output, <R as Reduce>::Error>
29where29where
@@ -32,8 +32,6 @@ where · 2 unmodified lines
32 O: Send,32 O: Send,
33{33{
34 let num_threads = num_threads(thread_limit);34 let num_threads = num_threads(thread_limit);
35 let new_thread_state = &new_thread_state;
36 let consume = &consume;
37 crossbeam_utils::thread::scope(move |s| {35 crossbeam_utils::thread::scope(move |s| {
38 let receive_result = {36 let receive_result = {
39 let (send_input, receive_input) = crossbeam_channel::bounded::<I>(num_threads);37 let (send_input, receive_input) = crossbeam_channel::bounded::<I>(num_threads);
@@ -42,6 +40,8 @@ where · 2 unmodified lines
42 s.spawn({40 s.spawn({
43 let send_result = send_result.clone();41 let send_result = send_result.clone();
44 let receive_input = receive_input.clone();42 let receive_input = receive_input.clone();
43 let new_thread_state = new_thread_state.clone();
44 let consume = consume.clone();
45 move |_| {45 move |_| {
46 let mut state = new_thread_state(thread_id);46 let mut state = new_thread_state(thread_id);
47 for item in receive_input {47 for item in receive_input {
git-features/src/parallel/mod.rs+2−2
@@ -133,8 +133,8 @@ pub fn in_parallel_if<I, S, O, R>( · 132 unmodified lines
133 condition: impl FnOnce() -> bool,133 condition: impl FnOnce() -> bool,
134 input: impl Iterator<Item = I> + Send,134 input: impl Iterator<Item = I> + Send,
135 thread_limit: Option<usize>,135 thread_limit: Option<usize>,
136 new_thread_state: impl Fn(usize) -> S + Send + Sync,136 new_thread_state: impl Fn(usize) -> S + Send + Clone,
137 consume: impl Fn(I, &mut S) -> O + Send + Sync,137 consume: impl Fn(I, &mut S) -> O + Send + Clone,
138 reducer: R,138 reducer: R,
139) -> Result<<R as Reduce>::Output, <R as Reduce>::Error>139) -> Result<<R as Reduce>::Output, <R as Reduce>::Error>
140where140where
git-features/src/parallel/reduce.rs+7−13
@@ -138,11 +138,9 @@ mod stepped { · 137 unmodified lines
138 138
139 impl<InputIter, ConsumeFn, Reduce, I, O, S> Stepwise<InputIter, ConsumeFn, S, Reduce>139 impl<InputIter, ConsumeFn, Reduce, I, O, S> Stepwise<InputIter, ConsumeFn, S, Reduce>
140 where140 where
141 InputIter: Iterator<Item = I> + Send,141 InputIter: Iterator<Item = I>,
142 ConsumeFn: Fn(I, &mut S) -> O + Send + Sync,142 ConsumeFn: Fn(I, &mut S) -> O,
143 Reduce: super::Reduce<Input = O>,143 Reduce: super::Reduce<Input = O>,
144 I: Send,
145 O: Send,
146 {144 {
147 /// Instantiate a new iterator.145 /// Instantiate a new iterator.
148 /// For a description of parameters, see [`in_parallel()`][crate::parallel::in_parallel()].146 /// For a description of parameters, see [`in_parallel()`][crate::parallel::in_parallel()].
@@ -154,7 +152,7 @@ mod stepped { · 5 unmodified lines
154 reducer: Reduce,152 reducer: Reduce,
155 ) -> Self153 ) -> Self
156 where154 where
157 ThreadStateFn: Fn(usize) -> S + Send + Sync,155 ThreadStateFn: Fn(usize) -> S,
158 {156 {
159 Stepwise {157 Stepwise {
160 input,158 input,
@@ -175,11 +173,9 @@ mod stepped { · 14 unmodified lines
175 173
176 impl<InputIter, ConsumeFn, ThreadState, Reduce, I, O> Iterator for Stepwise<InputIter, ConsumeFn, ThreadState, Reduce>174 impl<InputIter, ConsumeFn, ThreadState, Reduce, I, O> Iterator for Stepwise<InputIter, ConsumeFn, ThreadState, Reduce>
177 where175 where
178 InputIter: Iterator<Item = I> + Send,176 InputIter: Iterator<Item = I>,
179 ConsumeFn: Fn(I, &mut ThreadState) -> O + Send + Sync,177 ConsumeFn: Fn(I, &mut ThreadState) -> O,
180 Reduce: super::Reduce<Input = O>,178 Reduce: super::Reduce<Input = O>,
181 I: Send,
182 O: Send,
183 {179 {
184 type Item = Result<Reduce::FeedProduce, Reduce::Error>;180 type Item = Result<Reduce::FeedProduce, Reduce::Error>;
185 181
@@ -192,11 +188,9 @@ mod stepped { · 6 unmodified lines
192 188
193 impl<InputIter, ConsumeFn, R, I, O, S> super::Finalize for Stepwise<InputIter, ConsumeFn, S, R>189 impl<InputIter, ConsumeFn, R, I, O, S> super::Finalize for Stepwise<InputIter, ConsumeFn, S, R>
194 where190 where
195 InputIter: Iterator<Item = I> + Send,191 InputIter: Iterator<Item = I>,
196 ConsumeFn: Fn(I, &mut S) -> O + Send + Sync,192 ConsumeFn: Fn(I, &mut S) -> O,
197 R: super::Reduce<Input = O>,193 R: super::Reduce<Input = O>,
198 I: Send,
199 O: Send,
200 {194 {
201 type Reduce = R;195 type Reduce = R;
202 196
git-pack/src/data/output/count/objects/mod.rs+2−2
@@ -39,7 +39,7 @@ pub type Result<E1, E2> = std::result::Result<(Vec<output::Count>, Outcome), Err · 38 unmodified lines
39/// * more configuration39/// * more configuration
40pub fn objects<Find, Iter, IterErr, Oid, PackCache, ObjectCache>(40pub fn objects<Find, Iter, IterErr, Oid, PackCache, ObjectCache>(
41 db: Find,41 db: Find,
42 make_caches: impl Fn() -> (PackCache, ObjectCache) + Send + Sync,42 make_caches: impl Fn() -> (PackCache, ObjectCache) + Send + Clone,
43 objects_ids: Iter,43 objects_ids: Iter,
44 progress: impl Progress,44 progress: impl Progress,
45 should_interrupt: &AtomicBool,45 should_interrupt: &AtomicBool,
@@ -50,7 +50,7 @@ pub fn objects<Find, Iter, IterErr, Oid, PackCache, ObjectCache>( · 4 unmodified lines
50 }: Options,50 }: Options,
51) -> Result<find::existing::Error<Find::Error>, IterErr>51) -> Result<find::existing::Error<Find::Error>, IterErr>
52where52where
53 Find: crate::Find + Send + Sync,53 Find: crate::Find + Send + Clone,
54 <Find as crate::Find>::Error: Send,54 <Find as crate::Find>::Error: Send,
55 Iter: Iterator<Item = std::result::Result<Oid, IterErr>> + Send,55 Iter: Iterator<Item = std::result::Result<Oid, IterErr>> + Send,
56 Oid: Into<ObjectId> + Send,56 Oid: Into<ObjectId> + Send,
git-pack/src/data/output/entry/iter_from_counts.rs+3−3
@@ -42,7 +42,7 @@ use crate::data::{output, output::ChunkId}; · 41 unmodified lines
42pub fn iter_from_counts<Find, Cache>(42pub fn iter_from_counts<Find, Cache>(
43 mut counts: Vec<output::Count>,43 mut counts: Vec<output::Count>,
44 db: Find,44 db: Find,
45 make_cache: impl Fn() -> Cache + Send + Sync + 'static,45 make_cache: impl Fn() -> Cache + Send + Clone + Sync + 'static,
46 mut progress: impl Progress,46 mut progress: impl Progress,
47 Options {47 Options {
48 version,48 version,
@@ -54,7 +54,7 @@ pub fn iter_from_counts<Find, Cache>( · 5 unmodified lines
54) -> impl Iterator<Item = Result<(ChunkId, Vec<output::Entry>), Error<Find::Error>>>54) -> impl Iterator<Item = Result<(ChunkId, Vec<output::Entry>), Error<Find::Error>>>
55 + parallel::reduce::Finalize<Reduce = reduce::Statistics<Error<Find::Error>>>55 + parallel::reduce::Finalize<Reduce = reduce::Statistics<Error<Find::Error>>>
56where56where
57 Find: crate::Find + Send + Sync + 'static,57 Find: crate::Find + Send + Clone + 'static,
58 <Find as crate::Find>::Error: Send,58 <Find as crate::Find>::Error: Send,
59 Cache: crate::cache::DecodeEntry,59 Cache: crate::cache::DecodeEntry,
60{60{
@@ -78,7 +78,7 @@ where · 17 unmodified lines
78 {78 {
79 let progress = Arc::clone(&progress);79 let progress = Arc::clone(&progress);
80 let counts = &counts;80 let counts = &counts;
81 let db = &db;81 let db = db.clone();
82 move |chunk_range, buf| {82 move |chunk_range, buf| {
83 let chunk = {83 let chunk = {
84 let c = &counts[chunk_range];84 let c = &counts[chunk_range];

Keyboard shortcuts

General

Show keyboard shortcuts
?
Focus the current search
/

Issues and pull requests

Open filters
f

Issues, pull requests and Inbox

Select next item
j↓
Select previous item
k↑
Open selected item
Enter
Dismiss filters or clear selection
Esc

Repository code

Go to file
t