Add best-effort stealing API to PollingSystem

core/jvm-native/src/main/scala/cats/effect/unsafe/PollingSystem.scala

@@ -91,6 +91,23 @@ abstract class PollingSystem {
    */
   def poll(poller: Poller, nanos: Long, reportFailure: Throwable => Unit): Boolean
 
+  /**
+   * Makes a best-effort to steal completed I/O events. Not all polling systems support this.
+   *
+   * This method is safe to call concurrently from threads that do not own the poller.
+   *
+   * @param poller
+   *   the thread-local [[Poller]] used to poll events.
+   *
+   * @param reportFailure
+   *   callback that handles any failures that occur during stealing.
+   *
+   * @return
+   *   whether any events were stolen. e.g. if the method returned due to timeout, this should
+   *   be `false`.
+   */
+  def steal(poller: Poller, reportFailure: Throwable => Unit): Boolean
+
   /**
    * @return
    *   whether poll should be called again (i.e., there are more events to be polled)

core/jvm/src/main/scala/cats/effect/unsafe/SelectorSystem.scala

@@ -63,7 +63,7 @@ final class SelectorSystem private (provider: SelectorProvider) extends PollingS
         } catch {
           case ex if NonFatal(ex) =>
             error = ex
-            readyOps = -1 // interest all waiters
+            readyOps = -1 // notify all waiters
         }
 
         val value = if (error ne null) Left(error) else Right(readyOps)
@@ -98,6 +98,8 @@ final class SelectorSystem private (provider: SelectorProvider) extends PollingS
     } else false
   }
 
+  def steal(poller: Poller, reportFailure: Throwable => Unit): Boolean = false
+
   def needsPoll(poller: Poller): Boolean =
     !poller.selector.keys().isEmpty()
 

core/jvm/src/main/scala/cats/effect/unsafe/SleepSystem.scala

@@ -42,6 +42,8 @@ object SleepSystem extends PollingSystem {
     false
   }
 
+  def steal(poller: Poller, reportFailure: Throwable => Unit): Boolean = false
+
   def needsPoll(poller: Poller): Boolean = false
 
   def interrupt(targetThread: Thread, targetPoller: Poller): Unit =

core/native/src/main/scala/cats/effect/unsafe/EpollSystem.scala

@@ -62,6 +62,8 @@ object EpollSystem extends PollingSystem {
   def poll(poller: Poller, nanos: Long, reportFailure: Throwable => Unit): Boolean =
     poller.poll(nanos)
 
+  def steal(poller: Poller, reportFailure: Throwable => Unit): Boolean = false
+
   def needsPoll(poller: Poller): Boolean = poller.needsPoll()
 
   def interrupt(targetThread: Thread, targetPoller: Poller): Unit = ()

core/native/src/main/scala/cats/effect/unsafe/KqueueSystem.scala

@@ -61,6 +61,8 @@ object KqueueSystem extends PollingSystem {
   def poll(poller: Poller, nanos: Long, reportFailure: Throwable => Unit): Boolean =
     poller.poll(nanos)
 
+  def steal(poller: Poller, reportFailure: Throwable => Unit): Boolean = false
+
   def needsPoll(poller: Poller): Boolean =
     poller.needsPoll()
 

core/native/src/main/scala/cats/effect/unsafe/PollingExecutorScheduler.scala

@@ -43,6 +43,7 @@ abstract class PollingExecutorScheduler(pollEvery: Int)
             poller.poll(nanos.nanos)
         true
       }
+      def steal(poller: Poller, reportFailure: Throwable => Unit): Boolean = false
       def needsPoll(poller: Poller) = needsPoll
       def interrupt(targetThread: Thread, targetPoller: Poller): Unit = ()
     }

core/native/src/main/scala/cats/effect/unsafe/SleepSystem.scala

@@ -36,6 +36,8 @@ object SleepSystem extends PollingSystem {
     false
   }
 
+  def steal(poller: Poller, reportFailure: Throwable => Unit): Boolean = false
+
   def needsPoll(poller: Poller): Boolean = false
 
   def interrupt(targetThread: Thread, targetPoller: Poller): Unit = ()

tests/jvm/src/test/scala/cats/effect/IOPlatformSpecification.scala

@@ -513,6 +513,8 @@ trait IOPlatformSpecification extends DetectPlatform { self: BaseSpec with Scala
             }
           }
 
+          def steal(poller: Poller, reportFailure: Throwable => Unit) = false
+
           def makeApi(access: (Poller => Unit) => Unit): DummySystem.Api =
             new DummyPoller {
               def poll = IO.async_[Unit] { cb =>