Use CE polling system for non-blocking I/O on Native

build.sbt

@@ -2,7 +2,7 @@ import com.typesafe.tools.mima.core._
 
 Global / onChangedBuildSource := ReloadOnSourceChanges
 
-ThisBuild / tlBaseVersion := "3.4"
+ThisBuild / tlBaseVersion := "3.5"
 
 ThisBuild / organization := "co.fs2"
 ThisBuild / organizationName := "Functional Streams for Scala"
@@ -209,9 +209,9 @@ lazy val core = crossProject(JVMPlatform, JSPlatform, NativePlatform)
     libraryDependencies ++= Seq(
       "org.typelevel" %%% "cats-core" % "2.9.0",
       "org.typelevel" %%% "cats-laws" % "2.9.0" % Test,
-      "org.typelevel" %%% "cats-effect" % "3.4.2",
-      "org.typelevel" %%% "cats-effect-laws" % "3.4.2" % Test,
-      "org.typelevel" %%% "cats-effect-testkit" % "3.4.2" % Test,
+      "org.typelevel" %%% "cats-effect" % "3.5-9ba870f",
+      "org.typelevel" %%% "cats-effect-laws" % "3.5-9ba870f" % Test,
+      "org.typelevel" %%% "cats-effect-testkit" % "3.5-9ba870f" % Test,
       "org.scodec" %%% "scodec-bits" % "1.1.34",
       "org.typelevel" %%% "scalacheck-effect-munit" % "2.0.0-M2" % Test,
       "org.typelevel" %%% "munit-cats-effect" % "2.0.0-M3" % Test,

io/jvm-native/src/main/scala/fs2/io/iojvmnative.scala

@@ -22,46 +22,16 @@
 package fs2
 package io
 
-import cats._
 import cats.effect.kernel.Sync
 import cats.effect.kernel.implicits._
 import cats.syntax.all._
 
-import java.nio.charset.Charset
-import java.nio.charset.StandardCharsets
 import scala.reflect.ClassTag
 
 private[fs2] trait iojvmnative {
   type InterruptedIOException = java.io.InterruptedIOException
   type ClosedChannelException = java.nio.channels.ClosedChannelException
 
-  //
-  // STDIN/STDOUT Helpers
-
-  /** Stream of bytes read asynchronously from standard input. */
-  def stdin[F[_]: Sync](bufSize: Int): Stream[F, Byte] =
-    readInputStream(Sync[F].blocking(System.in), bufSize, false)
-
-  /** Pipe of bytes that writes emitted values to standard output asynchronously. */
-  def stdout[F[_]: Sync]: Pipe[F, Byte, Nothing] =
-    writeOutputStream(Sync[F].blocking(System.out), false)
-
-  /** Pipe of bytes that writes emitted values to standard error asynchronously. */
-  def stderr[F[_]: Sync]: Pipe[F, Byte, Nothing] =
-    writeOutputStream(Sync[F].blocking(System.err), false)
-
-  /** Writes this stream to standard output asynchronously, converting each element to
-    * a sequence of bytes via `Show` and the given `Charset`.
-    */
-  def stdoutLines[F[_]: Sync, O: Show](
-      charset: Charset = StandardCharsets.UTF_8
-  ): Pipe[F, O, Nothing] =
-    _.map(_.show).through(text.encode(charset)).through(stdout)
-
-  /** Stream of `String` read asynchronously from standard input decoded in UTF-8. */
-  def stdinUtf8[F[_]: Sync](bufSize: Int): Stream[F, String] =
-    stdin(bufSize).through(text.utf8.decode)
-
   /** Stream of bytes read asynchronously from the specified resource relative to the class `C`.
     * @see [[readClassLoaderResource]] for a resource relative to a classloader.
     */

io/jvm/src/main/scala/fs2/io/ioplatform.scala

@@ -22,16 +22,46 @@
 package fs2
 package io
 
+import cats.Show
 import cats.effect.kernel.{Async, Outcome, Resource, Sync}
 import cats.effect.kernel.implicits._
 import cats.effect.kernel.Deferred
 import cats.syntax.all._
 import fs2.io.internal.PipedStreamBuffer
 
 import java.io.{InputStream, OutputStream}
+import java.nio.charset.Charset
+import java.nio.charset.StandardCharsets
 
 private[fs2] trait ioplatform extends iojvmnative {
 
+  //
+  // STDIN/STDOUT Helpers
+
+  /** Stream of bytes read asynchronously from standard input. */
+  def stdin[F[_]: Sync](bufSize: Int): Stream[F, Byte] =
+    readInputStream(Sync[F].blocking(System.in), bufSize, false)
+
+  /** Pipe of bytes that writes emitted values to standard output asynchronously. */
+  def stdout[F[_]: Sync]: Pipe[F, Byte, Nothing] =
+    writeOutputStream(Sync[F].blocking(System.out), false)
+
+  /** Pipe of bytes that writes emitted values to standard error asynchronously. */
+  def stderr[F[_]: Sync]: Pipe[F, Byte, Nothing] =
+    writeOutputStream(Sync[F].blocking(System.err), false)
+
+  /** Writes this stream to standard output asynchronously, converting each element to
+    * a sequence of bytes via `Show` and the given `Charset`.
+    */
+  def stdoutLines[F[_]: Sync, O: Show](
+      charset: Charset = StandardCharsets.UTF_8
+  ): Pipe[F, O, Nothing] =
+    _.map(_.show).through(text.encode(charset)).through(stdout)
+
+  /** Stream of `String` read asynchronously from standard input decoded in UTF-8. */
+  def stdinUtf8[F[_]: Sync](bufSize: Int): Stream[F, String] =
+    stdin(bufSize).through(text.utf8.decode)
+
   /** Pipe that converts a stream of bytes to a stream that will emit a single `java.io.InputStream`,
     * that is closed whenever the resulting stream terminates.
     *

io/native/src/main/scala/fs2/io/internal/NativeUtil.scala

@@ -0,0 +1,79 @@
+/*
+ * Copyright (c) 2013 Functional Streams for Scala
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to
+ * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+ * the Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+ * IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+ * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+
+package fs2.io.internal
+
+import cats.effect.Sync
+
+import java.io.IOException
+import java.net.BindException
+import java.net.ConnectException
+import scala.scalanative.annotation.alwaysinline
+import scala.scalanative.libc.errno._
+import scala.scalanative.posix.fcntl._
+import scala.scalanative.posix.errno._
+import scala.scalanative.posix.string._
+import scala.scalanative.unsafe._
+
+private[io] object NativeUtil {
+
+  @alwaysinline def guard_(thunk: => CInt): Unit = {
+    guard(thunk)
+    ()
+  }
+
+  @alwaysinline def guard(thunk: => CInt): CInt = {
+    val rtn = thunk
+    if (rtn < 0) {
+      val e = errno
+      if (e == EAGAIN || e == EWOULDBLOCK)
+        rtn
+      else throw errnoToThrowable(e)
+    } else
+      rtn
+  }
+
+  @alwaysinline def guardSSize(thunk: => CSSize): CSSize = {
+    val rtn = thunk
+    if (rtn < 0) {
+      val e = errno
+      if (e == EAGAIN || e == EWOULDBLOCK)
+        rtn
+      else throw errnoToThrowable(e)
+    } else
+      rtn
+  }
+
+  @alwaysinline def errnoToThrowable(e: CInt): Throwable = {
+    val msg = fromCString(strerror(e))
+    if (e == EADDRINUSE /* || e == EADDRNOTAVAIL */ )
+      new BindException(msg)
+    else if (e == ECONNREFUSED)
+      new ConnectException(msg)
+    else
+      new IOException(msg)
+  }
+
+  def setNonBlocking[F[_]](fd: CInt)(implicit F: Sync[F]): F[Unit] = F.delay {
+    guard_(fcntl(fd, F_SETFL, O_NONBLOCK))
+  }
+
+}

io/native/src/main/scala/fs2/io/internal/ResizableBuffer.scala

@@ -21,44 +21,52 @@
 
 package fs2.io.internal
 
+import cats.effect.kernel.Async
 import cats.effect.kernel.Resource
-import cats.effect.kernel.Sync
+import cats.effect.std.Semaphore
 import cats.syntax.all._
 
 import scala.scalanative.libc.errno._
 import scala.scalanative.libc.stdlib._
+import scala.scalanative.posix.string._
 import scala.scalanative.unsafe._
 import scala.scalanative.unsigned._
 
 private[io] final class ResizableBuffer[F[_]] private (
+    semaphore: Semaphore[F],
     private var ptr: Ptr[Byte],
     private[this] var size: Int
-)(implicit F: Sync[F]) {
+)(implicit F: Async[F]) {
 
-  def get(size: Int): F[Ptr[Byte]] = F.delay {
-    if (size <= this.size)
-      F.pure(ptr)
-    else {
-      ptr = realloc(ptr, size.toUInt)
-      this.size = size
-      if (ptr == null)
-        F.raiseError[Ptr[Byte]](new RuntimeException(s"realloc: ${errno}"))
-      else F.pure(ptr)
+  def get(size: Int): Resource[F, Ptr[Byte]] = semaphore.permit.evalMap { _ =>
+    F.delay {
+      if (size <= this.size)
+        ptr
+      else {
+        ptr = realloc(ptr, size.toUInt)
+        this.size = size
+        if (ptr == null)
+          throw new RuntimeException(fromCString(strerror(errno)))
+        else ptr
+      }
     }
-  }.flatten
+  }
 
 }
 
 private[io] object ResizableBuffer {
 
-  def apply[F[_]](size: Int)(implicit F: Sync[F]): Resource[F, ResizableBuffer[F]] =
+  def apply[F[_]](size: Int)(implicit F: Async[F]): Resource[F, ResizableBuffer[F]] =
     Resource.make {
-      F.delay {
-        val ptr = malloc(size.toUInt)
-        if (ptr == null)
-          throw new RuntimeException(s"malloc: ${errno}")
-        else new ResizableBuffer(ptr, size)
+      Semaphore[F](1).flatMap { semaphore =>
+        F.delay {
+          val ptr = malloc(size.toUInt)
+          if (ptr == null)
+            throw new RuntimeException(fromCString(strerror(errno)))
+          else new ResizableBuffer(semaphore, ptr, size)
+        }
       }
+
     }(buf => F.delay(free(buf.ptr)))
 
 }