@@ -3,7 +3,7 @@ use std::sync::{Arc, atomic::AtomicUsize, atomic::Ordering};
33use std:: task:: { Context , Poll } ;
44use std:: { any:: Any , fmt, future:: Future , panic, pin:: Pin , thread, time:: Duration } ;
55
6- use crossbeam_channel:: { Receiver , Sender , TrySendError , bounded, unbounded} ;
6+ use crossbeam_channel:: { Receiver , Select , Sender , TrySendError , bounded, unbounded} ;
77
88/// An error that may be emitted when all worker threads are busy.
99#[ derive( Copy , Clone , Debug , PartialEq , Eq ) ]
@@ -19,7 +19,7 @@ impl fmt::Display for BlockingError {
1919
2020#[ derive( Debug ) ]
2121pub struct BlockingResult < T > {
22- rx : Option < oneshot:: AsyncReceiver < Result < T , Box < dyn Any + Send > > > > ,
22+ rx : oneshot:: AsyncReceiver < Result < T , Box < dyn Any + Send > > > ,
2323}
2424
2525type BoxedDispatchable = Box < dyn Dispatchable + Send > ;
@@ -46,15 +46,31 @@ impl Drop for CounterGuard {
4646}
4747
4848fn worker (
49- receiver : Receiver < BoxedDispatchable > ,
49+ receiver_high_prio : Receiver < BoxedDispatchable > ,
50+ receiver_low_prio : Receiver < BoxedDispatchable > ,
5051 counter : Arc < AtomicUsize > ,
5152 timeout : Duration ,
5253) -> impl FnOnce ( ) {
5354 move || {
5455 counter. fetch_add ( 1 , Ordering :: AcqRel ) ;
5556 let _guard = CounterGuard ( counter) ;
56- while let Ok ( f) = receiver. recv_timeout ( timeout) {
57- f. run ( ) ;
57+ let mut sel = Select :: new_biased ( ) ;
58+ sel. recv ( & receiver_high_prio) ;
59+ sel. recv ( & receiver_low_prio) ;
60+ while let Ok ( op) = sel. select_timeout ( timeout) {
61+ match op {
62+ op if op. index ( ) == 0 => {
63+ if let Ok ( f) = op. recv ( & receiver_high_prio) {
64+ f. run ( ) ;
65+ }
66+ }
67+ op if op. index ( ) == 1 => {
68+ if let Ok ( f) = op. recv ( & receiver_low_prio) {
69+ f. run ( ) ;
70+ }
71+ }
72+ _ => unreachable ! ( ) ,
73+ }
5874 }
5975 }
6076}
@@ -74,8 +90,10 @@ fn worker(
7490#[ derive( Debug , Clone ) ]
7591pub struct ThreadPool {
7692 name : String ,
77- sender : Sender < BoxedDispatchable > ,
78- receiver : Receiver < BoxedDispatchable > ,
93+ sender_low_prio : Sender < BoxedDispatchable > ,
94+ receiver_low_prio : Receiver < BoxedDispatchable > ,
95+ sender_high_prio : Sender < BoxedDispatchable > ,
96+ receiver_high_prio : Receiver < BoxedDispatchable > ,
7997 counter : Arc < AtomicUsize > ,
8098 thread_limit : usize ,
8199 recv_timeout : Duration ,
@@ -84,16 +102,14 @@ pub struct ThreadPool {
84102impl ThreadPool {
85103 /// Creates a [`ThreadPool`] with a maximum number of worker threads
86104/// and a timeout for receiving tasks from the task channel.
87- pub fn new (
88- name : & str ,
89- thread_limit : usize ,
90- recv_timeout : Duration ,
91- bound : bool ,
92- ) -> Self {
93- let ( sender, receiver) = if bound { bounded ( 0 ) } else { unbounded ( ) } ;
105+ pub fn new ( name : & str , thread_limit : usize , recv_timeout : Duration ) -> Self {
106+ let ( sender_low_prio, receiver_low_prio) = bounded ( 0 ) ;
107+ let ( sender_high_prio, receiver_high_prio) = unbounded ( ) ;
94108 Self {
95- sender,
96- receiver,
109+ sender_low_prio,
110+ receiver_low_prio,
111+ sender_high_prio,
112+ receiver_high_prio,
97113 thread_limit,
98114 recv_timeout,
99115 name : format ! ( "{name}:pool-wrk" ) ,
@@ -106,43 +122,45 @@ impl ThreadPool {
106122///
107123/// The task will be executed by an available worker thread.
108124/// If no threads are available and the pool has reached its maximum size,
109- /// the behavior depends on the `boundedness` configuration:
110- ///
111- /// - For a bounded pool, the function returns an error.
112- /// - For an unbounded pool, the task is queued and executed when a worker
113- /// becomes available.
125+ /// the work will be queued until a worker thread becomes available.
114126pub fn execute < F , R > ( & self , f : F ) -> BlockingResult < R >
115127 where
116128 F : FnOnce ( ) -> R + Send + ' static ,
117129 R : Send + ' static ,
118130 {
119131 let ( tx, rx) = oneshot:: async_channel ( ) ;
120132 let f = Box :: new ( move || {
121- // do not execute operation if recevier is dropped
133+ // do not execute operation if receiver is dropped
122134 if !tx. is_closed ( ) {
123135 let result = panic:: catch_unwind ( panic:: AssertUnwindSafe ( f) ) ;
124136 let _ = tx. send ( result) ;
125137 }
126138 } ) ;
127139
128- match self . sender . try_send ( f) {
129- Ok ( ( ) ) => BlockingResult { rx : Some ( rx ) } ,
140+ match self . sender_low_prio . try_send ( f) {
141+ Ok ( ( ) ) => BlockingResult { rx } ,
130142 Err ( e) => match e {
131143 TrySendError :: Full ( f) => {
132144 let cnt = self . counter . load ( Ordering :: Acquire ) ;
133145 if cnt >= self . thread_limit {
134- BlockingResult { rx : None }
146+ self . sender_high_prio
147+ . send ( f)
148+ . expect ( "the channel should not be full" ) ;
149+ BlockingResult { rx }
135150 } else {
136151 thread:: Builder :: new ( )
137152 . name ( format ! ( "{}:{}" , self . name, cnt) )
138153 . spawn ( worker (
139- self . receiver . clone ( ) ,
154+ self . receiver_high_prio . clone ( ) ,
155+ self . receiver_low_prio . clone ( ) ,
140156 self . counter . clone ( ) ,
141157 self . recv_timeout ,
142158 ) )
143159 . expect ( "Cannot construct new thread" ) ;
144- self . sender . send ( f) . expect ( "the channel should not be full" ) ;
145- BlockingResult { rx : Some ( rx) }
160+ self . sender_low_prio
161+ . send ( f)
162+ . expect ( "the channel should not be full" ) ;
163+ BlockingResult { rx }
146164 }
147165 }
148166 TrySendError :: Disconnected ( _) => {
@@ -159,24 +177,13 @@ impl<R> Future for BlockingResult<R> {
159177 fn poll ( self : Pin < & mut Self > , cx : & mut Context < ' _ > ) -> Poll < Self :: Output > {
160178 let this = self . get_mut ( ) ;
161179
162- if this. rx . is_none ( ) {
163- return Poll :: Ready ( Err ( BlockingError ) ) ;
164- }
165-
166- if let Some ( mut rx) = this. rx . take ( ) {
167- match Pin :: new ( & mut rx) . poll ( cx) {
168- Poll :: Pending => {
169- this. rx = Some ( rx) ;
170- Poll :: Pending
171- }
172- Poll :: Ready ( result) => Poll :: Ready (
173- result
174- . map_err ( |_| BlockingError )
175- . and_then ( |res| res. map_err ( |_| BlockingError ) ) ,
176- ) ,
177- }
178- } else {
179- unreachable ! ( )
180+ match Pin :: new ( & mut this. rx ) . poll ( cx) {
181+ Poll :: Pending => Poll :: Pending ,
182+ Poll :: Ready ( result) => Poll :: Ready (
183+ result
184+ . map_err ( |_| BlockingError )
185+ . and_then ( |res| res. map_err ( |_| BlockingError ) ) ,
186+ ) ,
180187 }
181188 }
182189}
0 commit comments