44//! and allow scaling to more cores.
55//! However, that would be more complex, so we use this as a starting point.
66
7- use std:: { mem, io , sync:: mpsc:: { self , Sender } } ;
7+ use std:: { mem, sync:: mpsc:: { self , Receiver , Sender } } ;
88use crate :: decoder:: MAX_COMPONENTS ;
99use crate :: error:: Result ;
1010use super :: { RowData , Worker } ;
1111use super :: immediate:: ImmediateWorker ;
1212
13+ pub fn with_multithreading < T > ( f : impl FnOnce ( & mut dyn Worker ) -> T ) -> T {
14+ #[ cfg( not( feature = "rayon" ) ) ]
15+ return self :: enter_threads ( f) ;
16+
17+ #[ cfg( feature = "rayon" ) ]
18+ return jpeg_rayon:: enter ( |mut worker| {
19+ f ( & mut worker)
20+ } ) ;
21+ }
22+
1323enum WorkerMsg {
1424 Start ( RowData ) ,
1525 AppendRow ( Vec < i16 > ) ,
1626 GetResult ( Sender < Vec < u8 > > ) ,
1727}
18- pub struct MultiThreadedWorker {
28+
29+ #[ derive( Default ) ]
30+ pub struct MpscWorker {
1931 senders : [ Option < Sender < WorkerMsg > > ; MAX_COMPONENTS ]
2032}
2133
22- impl Worker for MultiThreadedWorker {
23- fn new ( ) -> Result < Self > {
24- Ok ( MultiThreadedWorker {
25- senders : [ None , None , None , None ]
26- } )
27- }
28- fn start ( & mut self , row_data : RowData ) -> Result < ( ) > {
34+ pub struct StdThreadWorker ( MpscWorker ) ;
35+
36+ impl MpscWorker {
37+ fn start_with (
38+ & mut self ,
39+ row_data : RowData ,
40+ spawn_worker : impl FnOnce ( usize ) -> Result < Sender < WorkerMsg > > ,
41+ ) -> Result < ( ) > {
2942 // if there is no worker thread for this component yet, start one
3043 let component = row_data. index ;
3144 if let None = self . senders [ component] {
32- let sender = spawn_worker_thread ( component) ?;
45+ let sender = spawn_worker ( component) ?;
3346 self . senders [ component] = Some ( sender) ;
3447 }
48+
3549 // we do the "take out value and put it back in once we're done" dance here
3650 // and in all other message-passing methods because there's not that many rows
3751 // and this should be cheaper than spawning MAX_COMPONENTS many threads up front
@@ -40,25 +54,42 @@ impl Worker for MultiThreadedWorker {
4054 self . senders [ component] = Some ( sender) ;
4155 Ok ( ( ) )
4256 }
57+
4358 fn append_row ( & mut self , row : ( usize , Vec < i16 > ) ) -> Result < ( ) > {
4459 let component = row. 0 ;
4560 let sender = mem:: replace ( & mut self . senders [ component] , None ) . unwrap ( ) ;
4661 sender. send ( WorkerMsg :: AppendRow ( row. 1 ) ) . expect ( "jpeg-decoder worker thread error" ) ;
4762 self . senders [ component] = Some ( sender) ;
4863 Ok ( ( ) )
4964 }
50- fn get_result ( & mut self , index : usize ) -> Result < Vec < u8 > > {
65+
66+ fn get_result_with (
67+ & mut self ,
68+ index : usize ,
69+ collect : impl FnOnce ( Receiver < Vec < u8 > > ) -> Vec < u8 > ,
70+ ) -> Result < Vec < u8 > > {
5171 let ( tx, rx) = mpsc:: channel ( ) ;
5272 let sender = mem:: replace ( & mut self . senders [ index] , None ) . unwrap ( ) ;
5373 sender. send ( WorkerMsg :: GetResult ( tx) ) . expect ( "jpeg-decoder worker thread error" ) ;
54- Ok ( rx . recv ( ) . expect ( "jpeg-decoder worker thread error" ) )
74+ Ok ( collect ( rx ) )
5575 }
5676}
5777
58- fn spawn_worker_thread ( component : usize ) -> Result < Sender < WorkerMsg > > {
59- let ( tx, rx) = mpsc:: channel ( ) ;
78+ impl Worker for StdThreadWorker {
79+ fn start ( & mut self , row_data : RowData ) -> Result < ( ) > {
80+ self . 0 . start_with ( row_data, spawn_worker_thread)
81+ }
82+ fn append_row ( & mut self , row : ( usize , Vec < i16 > ) ) -> Result < ( ) > {
83+ self . 0 . append_row ( row)
84+ }
85+ fn get_result ( & mut self , index : usize ) -> Result < Vec < u8 > > {
86+ self . 0 . get_result_with ( index, collect_worker_thread)
87+ }
88+ }
6089
61- spawn ( component, move || {
90+ fn create_worker ( ) -> ( Sender < WorkerMsg > , impl FnOnce ( ) + ' static ) {
91+ let ( tx, rx) = mpsc:: channel ( ) ;
92+ let closure = move || {
6293 let mut worker = ImmediateWorker :: new_immediate ( ) ;
6394
6495 while let Ok ( message) = rx. recv ( ) {
@@ -79,27 +110,93 @@ fn spawn_worker_thread(component: usize) -> Result<Sender<WorkerMsg>> {
79110 } ,
80111 }
81112 }
82- } ) ?;
113+ } ;
114+
115+ ( tx, closure)
116+ }
83117
118+ fn spawn_worker_thread ( component : usize ) -> Result < Sender < WorkerMsg > > {
119+ let ( tx, worker) = create_worker ( ) ;
120+ let thread_builder =
121+ std:: thread:: Builder :: new ( ) . name ( format ! ( "worker thread for component {}" , component) ) ;
122+ thread_builder. spawn ( worker) ?;
84123 Ok ( tx)
85124}
86125
87- #[ cfg( feature = "rayon" ) ]
88- fn spawn < F > ( _component : usize , func : F ) -> io:: Result < ( ) >
89- where
90- F : FnOnce ( ) + Send + ' static ,
91- {
92- rayon:: spawn ( func) ;
93- Ok ( ( ) )
126+
127+ fn collect_worker_thread ( rx : Receiver < Vec < u8 > > ) -> Vec < u8 > {
128+ rx. recv ( ) . expect ( "jpeg-decoder worker thread error" )
94129}
95130
96- #[ cfg( not( feature = "rayon" ) ) ]
97- fn spawn < F > ( component : usize , func : F ) -> io:: Result < ( ) >
98- where
99- F : FnOnce ( ) + Send + ' static ,
100- {
101- let thread_builder =
102- std:: thread:: Builder :: new ( ) . name ( format ! ( "worker thread for component {}" , component) ) ;
103- thread_builder. spawn ( func) ?;
104- Ok ( ( ) )
131+ #[ allow( dead_code) ]
132+ fn enter_threads < T > ( f : impl FnOnce ( & mut dyn Worker ) -> T ) -> T {
133+ let mut worker = StdThreadWorker ( MpscWorker :: default ( ) ) ;
134+ f ( & mut worker)
105135}
136+
137+
138+ #[ cfg( feature = "rayon" ) ]
139+ mod jpeg_rayon {
140+ use crate :: error:: Result ;
141+ use super :: { MpscWorker , RowData } ;
142+
143+ pub struct Scoped < ' r , ' scope > {
144+ fifo : & ' r rayon:: ScopeFifo < ' scope > ,
145+ inner : MpscWorker ,
146+ }
147+
148+ pub fn enter < T > ( f : impl FnOnce ( Scoped ) -> T ) -> T {
149+ // Note: Must be at least two threads. Otherwise, we may deadlock, due to ordering
150+ // constraints that we can not impose properly. Note that `append_row` creates a new task
151+ // while in `get_result` we wait for all tasks of a component. The only way for rayon to
152+ // impose this wait __and get a result__ is by ending an in_place_scope.
153+ //
154+ // However, the ordering of tasks is not as FIFO as the name would suggest. Indeed, even
155+ // though tasks are spawned in `start` _before_ the task spawned in `get_result`, the
156+ // `in_place_scope_fifo` will wait for ITS OWN results in fifo order. This implies, unless
157+ // there is some other thread capable of stealing the worker the work task will in fact not
158+ // get executed and the result will wait forever. It is impossible to otherwise schedule
159+ // the worker tasks specifically (e.g. join handle would be cool *cough* if you read this
160+ // and work on rayon) before while yielding from the current thread.
161+ //
162+ // So: we need at least one more worker thread that is _not_ occupied.
163+ let threads = rayon:: ThreadPoolBuilder :: new ( ) . num_threads ( 4 ) . build ( ) . unwrap ( ) ;
164+
165+ threads. in_place_scope_fifo ( |fifo| {
166+ f ( Scoped { fifo, inner : MpscWorker :: default ( ) } )
167+ } )
168+ }
169+
170+ impl super :: Worker for Scoped < ' _ , ' _ > {
171+ fn start ( & mut self , row_data : RowData ) -> Result < ( ) > {
172+ let fifo = & mut self . fifo ;
173+ self . inner . start_with ( row_data, |_| {
174+ let ( tx, worker) = super :: create_worker ( ) ;
175+ fifo. spawn_fifo ( move |_| {
176+ worker ( )
177+ } ) ;
178+ Ok ( tx)
179+ } )
180+ }
181+
182+ fn append_row ( & mut self , row : ( usize , Vec < i16 > ) ) -> Result < ( ) > {
183+ self . inner . append_row ( row)
184+ }
185+
186+ fn get_result ( & mut self , index : usize ) -> Result < Vec < u8 > > {
187+ self . inner . get_result_with ( index, |rx| {
188+ let mut result = vec ! [ ] ;
189+ let deliver_result = & mut result;
190+
191+ rayon:: in_place_scope_fifo ( |scope| {
192+ scope. spawn_fifo ( move |_| {
193+ * deliver_result = rx. recv ( ) . expect ( "jpeg-decoder worker thread error" ) ;
194+ } ) ;
195+ } ) ;
196+
197+ result
198+ } )
199+ }
200+ }
201+ }
202+
0 commit comments