@@ -34,6 +34,11 @@ pub(crate) struct Reactor {
3434 poller : Poller ,
3535
3636 /// Ticker bumped before polling.
37+ ///
38+ /// This is useful for checking what is the current "round" of `ReactorLock::react()` when
39+ /// synchronizing things in `Source::readable()` and `Source::writable()`. Both of those
40+ /// methods must make sure they don't receive stale I/O events - they only accept events from a
41+ /// fresh "round" of `ReactorLock::react()`.
3742 ticker : AtomicUsize ,
3843
3944 /// Registered sources.
@@ -191,6 +196,9 @@ impl Reactor {
191196
192197 // Try grabbing a lock on the reactor to wait on I/O.
193198 if let Some ( mut reactor_lock) = Reactor :: get ( ) . try_lock ( ) {
199+ // Record the instant at which the lock was grabbed.
200+ let start = Instant :: now ( ) ;
201+ 194202 loop {
195203 // First let wakers know this parker is blocked on I/O.
196204 IO_POLLING . with ( |io| io. set ( true ) ) ;
@@ -200,7 +208,8 @@ impl Reactor {
200208 io_blocked. store ( false , Ordering :: SeqCst ) ;
201209 } ) ;
202210
203- // Check if a notification has been received.
211+ // Check if a notification has been received before `io_blocked` was updated
212+ // because in that case the reactor won't receive a wakeup.
204213 if p. park_timeout ( Duration :: from_secs ( 0 ) ) {
205214 break ;
206215 }
@@ -212,6 +221,23 @@ impl Reactor {
212221 if p. park_timeout ( Duration :: from_secs ( 0 ) ) {
213222 break ;
214223 }
224+ 225+ // Check if this thread been handling I/O events for a long time.
226+ if start. elapsed ( ) > Duration :: from_micros ( 500 ) {
227+ // This thread is clearly processing I/O events for some other threads
228+ // because it didn't get a notification yet. It's best to stop hogging the
229+ // reactor and give other threads a chance to process I/O events for
230+ // themselves.
231+ drop ( reactor_lock) ;
232+ 233+ // Unpark the "async-io" thread in case no other thread is ready to start
234+ // processing I/O events. This way we prevent a potential latency spike.
235+ self . thread_unparker . unpark ( ) ;
236+ 237+ // Wait for a notification.
238+ p. park ( ) ;
239+ break ;
240+ }
215241 }
216242 } else {
217243 // Wait for an actual notification.
0 commit comments