File tree Expand file tree Collapse file tree 2 files changed +15
-4
lines changed Expand file tree Collapse file tree 2 files changed +15
-4
lines changed Original file line number Diff line number Diff line change @@ -155,7 +155,9 @@ where
155
155
let alloc = crate :: communication:: allocator:: thread:: Thread :: new ( ) ;
156
156
let mut worker = crate :: worker:: Worker :: new ( WorkerConfig :: default ( ) , alloc) ;
157
157
let result = func ( & mut worker) ;
158
- while worker. step_or_park ( None ) { }
158
+ while worker. has_dataflows ( ) {
159
+ worker. step_or_park ( None ) ;
160
+ }
159
161
result
160
162
}
161
163
@@ -283,7 +285,9 @@ where
283
285
}
284
286
285
287
let result = func ( & mut worker) ;
286
- while worker. step_or_park ( None ) { }
288
+ while worker. has_dataflows ( ) {
289
+ worker. step_or_park ( None ) ;
290
+ }
287
291
result
288
292
} )
289
293
}
@@ -382,7 +386,9 @@ where
382
386
initialize_from ( builders, others, move |allocator| {
383
387
let mut worker = Worker :: new ( worker_config. clone ( ) , allocator) ;
384
388
let result = func ( & mut worker) ;
385
- while worker. step_or_park ( None ) { }
389
+ while worker. has_dataflows ( ) {
390
+ worker. step_or_park ( None ) ;
391
+ }
386
392
result
387
393
} )
388
394
}
Original file line number Diff line number Diff line change @@ -366,7 +366,7 @@ impl<A: Allocate> Worker<A> {
366
366
( x, y) => x. or ( y) ,
367
367
} ;
368
368
369
- if ! self . dataflows . borrow ( ) . is_empty ( ) && delay != Some ( Duration :: new ( 0 , 0 ) ) {
369
+ if delay != Some ( Duration :: new ( 0 , 0 ) ) {
370
370
371
371
// Log parking and flush log.
372
372
if let Some ( l) = self . logging ( ) . as_mut ( ) {
@@ -698,6 +698,11 @@ impl<A: Allocate> Worker<A> {
698
698
self . dataflows . borrow ( ) . keys ( ) . cloned ( ) . collect ( )
699
699
}
700
700
701
+ /// True if there is at least one dataflow under management.
702
+ pub fn has_dataflows ( & self ) -> bool {
703
+ !self . dataflows . borrow ( ) . is_empty ( )
704
+ }
705
+
701
706
// Acquire a new distinct dataflow identifier.
702
707
fn allocate_dataflow_index ( & mut self ) -> usize {
703
708
* self . dataflow_counter . borrow_mut ( ) += 1 ;
You can’t perform that action at this time.
0 commit comments