Native Polling System

build.sbt

@@ -772,6 +772,14 @@ lazy val core = crossProject(JSPlatform, JVMPlatform, NativePlatform)
       } else Seq()
     }
   )
+  .nativeSettings(
+    mimaBinaryIssueFilters ++= Seq(
+      ProblemFilters.exclude[MissingClassProblem](
+        "cats.effect.unsafe.PollingExecutorScheduler$SleepTask"),
+      ProblemFilters.exclude[MissingClassProblem]("cats.effect.unsafe.QueueExecutorScheduler"),
+      ProblemFilters.exclude[MissingClassProblem]("cats.effect.unsafe.QueueExecutorScheduler$")
+    )
+  )
 
 /**
  * Test support for the core project, providing various helpful instances like ScalaCheck

core/native/src/main/scala/cats/effect/IOApp.scala

@@ -20,6 +20,7 @@ import cats.effect.metrics.NativeCpuStarvationMetrics
 
 import scala.concurrent.CancellationException
 import scala.concurrent.duration._
+import scala.scalanative.meta.LinktimeInfo
 
 /**
  * The primary entry point to a Cats Effect application. Extend this trait rather than defining
@@ -165,6 +166,21 @@ trait IOApp {
    */
   protected def runtimeConfig: unsafe.IORuntimeConfig = unsafe.IORuntimeConfig()
 
+  /**
+   * The [[unsafe.PollingSystem]] used by the [[runtime]] which will evaluate the [[IO]]
+   * produced by `run`. It is very unlikely that users will need to override this method.
+   *
+   * [[unsafe.PollingSystem]] implementors may provide their own flavors of [[IOApp]] that
+   * override this method.
+   */
+  protected def pollingSystem: unsafe.PollingSystem =
+    if (LinktimeInfo.isLinux)
+      unsafe.EpollSystem(64)
+    else if (LinktimeInfo.isMac)
+      unsafe.KqueueSystem(64)
+    else
+      unsafe.SleepSystem
+
   /**
    * The entry point for your application. Will be called by the runtime when the process is
    * started. If the underlying runtime supports it, any arguments passed to the process will be
@@ -186,12 +202,8 @@ trait IOApp {
       import unsafe.IORuntime
 
       val installed = IORuntime installGlobal {
-        IORuntime(
-          IORuntime.defaultComputeExecutionContext,
-          IORuntime.defaultComputeExecutionContext,
-          IORuntime.defaultScheduler,
-          () => (),
-          runtimeConfig)
+        val loop = IORuntime.createEventLoop(pollingSystem)
+        IORuntime(loop, loop, loop, () => (), runtimeConfig)
       }
 
       _runtime = IORuntime.global

core/native/src/main/scala/cats/effect/IOCompanionPlatform.scala

@@ -17,6 +17,9 @@
 package cats.effect
 
 import cats.effect.std.Console
+import cats.effect.unsafe.EventLoop
+
+import scala.reflect.ClassTag
 
 import java.time.Instant
 
@@ -62,4 +65,11 @@ private[effect] abstract class IOCompanionPlatform { this: IO.type =>
    */
   def readLine: IO[String] =
     Console[IO].readLine
+
+  def eventLoop[Poller](implicit ct: ClassTag[Poller]): IO[Option[EventLoop[Poller]]] =
+    IO.executionContext.map {
+      case loop: EventLoop[_] if ct.runtimeClass.isInstance(loop.poller()) =>
+        Some(loop.asInstanceOf[EventLoop[Poller]])
+      case _ => None
+    }
 }

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

@@ -0,0 +1,159 @@
+/*
+ * Copyright 2020-2022 Typelevel
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package cats.effect
+package unsafe
+
+import scala.scalanative.libc.errno._
+import scala.scalanative.posix.string._
+import scala.scalanative.posix.unistd
+import scala.scalanative.unsafe._
+import scala.scalanative.unsigned._
+import scala.util.control.NonFatal
+
+import java.io.IOException
+import java.util.{Collections, IdentityHashMap, Set}
+
+import EpollSystem.epoll._
+import EpollSystem.epollImplicits._
+
+final class EpollSystem private (maxEvents: Int) extends PollingSystem {
+
+  def makePoller(): Poller = {
+    val fd = epoll_create1(0)
+    if (fd == -1)
+      throw new IOException(fromCString(strerror(errno)))
+    new Poller(fd, maxEvents)
+  }
+
+  def close(poller: Poller): Unit = poller.close()
+
+  def poll(poller: Poller, nanos: Long, reportFailure: Throwable => Unit): Boolean =
+    poller.poll(nanos, reportFailure)
+
+  final class Poller private[EpollSystem] (private[EpollSystem] val epfd: Int, maxEvents: Int)
+      extends FileDescriptorPoller {
+
+    private[this] val callbacks: Set[FileDescriptorPoller.Callback] =
+      Collections.newSetFromMap(new IdentityHashMap)
+
+    private[EpollSystem] def close(): Unit =
+      if (unistd.close(epfd) != 0)
+        throw new IOException(fromCString(strerror(errno)))
+
+    private[EpollSystem] def poll(timeout: Long, reportFailure: Throwable => Unit): Boolean = {
+      val noCallbacks = callbacks.isEmpty()
+
+      if (timeout <= 0 && noCallbacks)
+        false // nothing to do here
+      else {
+        val timeoutMillis = if (timeout == -1) -1 else (timeout / 1000000).toInt
+
+        val events = stackalloc[epoll_event](maxEvents.toUInt)
+
+        val triggeredEvents = epoll_wait(epfd, events, maxEvents, timeoutMillis)
+
+        if (triggeredEvents >= 0) {
+          var i = 0
+          while (i < triggeredEvents) {
+            val event = events + i.toLong
+            val cb = FileDescriptorPoller.Callback.fromPtr(event.data)
+            try {
+              val e = event.events.toInt
+              val readReady = (e & EPOLLIN) != 0
+              val writeReady = (e & EPOLLOUT) != 0
+              cb.notifyFileDescriptorEvents(readReady, writeReady)
+            } catch {
+              case ex if NonFatal(ex) => reportFailure(ex)
+            }
+            i += 1
+          }
+        } else {
+          throw new IOException(fromCString(strerror(errno)))
+        }
+
+        !callbacks.isEmpty()
+      }
+    }
+
+    def registerFileDescriptor(fd: Int, reads: Boolean, writes: Boolean)(
+        cb: FileDescriptorPoller.Callback): Runnable = {
+      val event = stackalloc[epoll_event]()
+      event.events =
+        (EPOLLET | (if (reads) EPOLLIN else 0) | (if (writes) EPOLLOUT else 0)).toUInt
+      event.data = FileDescriptorPoller.Callback.toPtr(cb)
+
+      if (epoll_ctl(epfd, EPOLL_CTL_ADD, fd, event) != 0)
+        throw new IOException(fromCString(strerror(errno)))
+      callbacks.add(cb)
+
+      () => {
+        callbacks.remove(cb)
+        if (epoll_ctl(epfd, EPOLL_CTL_DEL, fd, null) != 0)
+          throw new IOException(fromCString(strerror(errno)))
+      }
+    }
+  }
+
+}
+
+object EpollSystem {
+  def apply(maxEvents: Int): EpollSystem = new EpollSystem(maxEvents)
+
+  @extern
+  private[unsafe] object epoll {
+
+    final val EPOLL_CTL_ADD = 1
+    final val EPOLL_CTL_DEL = 2
+    final val EPOLL_CTL_MOD = 3
+
+    final val EPOLLIN = 0x001
+    final val EPOLLOUT = 0x004
+    final val EPOLLONESHOT = 1 << 30
+    final val EPOLLET = 1 << 31
+
+    type epoll_event
+    type epoll_data_t = Ptr[Byte]
+
+    def epoll_create1(flags: Int): Int = extern
+
+    def epoll_ctl(epfd: Int, op: Int, fd: Int, event: Ptr[epoll_event]): Int = extern
+
+    def epoll_wait(epfd: Int, events: Ptr[epoll_event], maxevents: Int, timeout: Int): Int =
+      extern
+
+  }
+
+  private[unsafe] object epollImplicits {
+
+    implicit final class epoll_eventOps(epoll_event: Ptr[epoll_event]) {
+      def events: CUnsignedInt = !(epoll_event.asInstanceOf[Ptr[CUnsignedInt]])
+      def events_=(events: CUnsignedInt): Unit =
+        !(epoll_event.asInstanceOf[Ptr[CUnsignedInt]]) = events
+
+      def data: epoll_data_t =
+        !((epoll_event.asInstanceOf[Ptr[Byte]] + sizeof[CUnsignedInt])
+          .asInstanceOf[Ptr[epoll_data_t]])
+      def data_=(data: epoll_data_t): Unit =
+        !((epoll_event.asInstanceOf[Ptr[Byte]] + sizeof[CUnsignedInt])
+          .asInstanceOf[Ptr[epoll_data_t]]) = data
+    }
+
+    implicit val epoll_eventTag: Tag[epoll_event] =
+      Tag.materializeCArrayTag[Byte, Nat.Digit2[Nat._1, Nat._2]].asInstanceOf[Tag[epoll_event]]
+
+  }
+}

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

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2020-2022 Typelevel
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package cats.effect
+package unsafe
+
+import scala.concurrent.ExecutionContext
+
+trait EventLoop[Poller] extends ExecutionContext {
+
+  def poller(): Poller
+
+}