I/O polling

build.sbt

@@ -299,9 +299,9 @@ lazy val core = crossProject(JVMPlatform, JSPlatform, NativePlatform)
     libraryDependencies ++= Seq(
       "org.scodec" %%% "scodec-bits" % "1.1.38",
       "org.typelevel" %%% "cats-core" % "2.11.0",
-      "org.typelevel" %%% "cats-effect" % "3.5.7",
-      "org.typelevel" %%% "cats-effect-laws" % "3.5.7" % Test,
-      "org.typelevel" %%% "cats-effect-testkit" % "3.5.7" % Test,
+      "org.typelevel" %%% "cats-effect" % "3.6.0-RC1",
+      "org.typelevel" %%% "cats-effect-laws" % "3.6.0-RC1" % Test,
+      "org.typelevel" %%% "cats-effect-testkit" % "3.6.0-RC1" % Test,
       "org.typelevel" %%% "cats-laws" % "2.11.0" % Test,
       "org.typelevel" %%% "discipline-munit" % "2.0.0-M3" % Test,
       "org.typelevel" %%% "munit-cats-effect" % "2.0.0" % Test,
@@ -370,9 +370,6 @@ lazy val io = crossProject(JVMPlatform, JSPlatform, NativePlatform)
   .nativeEnablePlugins(ScalaNativeBrewedConfigPlugin)
   .nativeSettings(commonNativeSettings)
   .nativeSettings(
-    libraryDependencies ++= Seq(
-      "com.armanbilge" %%% "epollcat" % "0.1.6" % Test
-    ),
     Test / nativeBrewFormulas += "s2n",
     Test / envVars ++= Map("S2N_DONT_MLOCK" -> "1")
   )

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

@@ -22,37 +22,15 @@
 package fs2
 package io
 
-import cats._
 import cats.effect.kernel.Sync
 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
-
-  /** 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 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,6 +22,7 @@
 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
@@ -30,6 +31,8 @@ import fs2.internal.ThreadFactories
 import fs2.io.internal.PipedStreamBuffer
 
 import java.io.{InputStream, OutputStream}
+import java.nio.charset.Charset
+import java.nio.charset.StandardCharsets
 import java.util.concurrent.Executors
 
 private[fs2] trait ioplatform extends iojvmnative {
@@ -71,6 +74,22 @@ private[fs2] trait ioplatform extends iojvmnative {
   def stdinUtf8[F[_]](bufSize: Int, F: Sync[F]): Stream[F, String] =
     stdin(bufSize, F).through(text.utf8.decode)
 
+  /** 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)
+
   /** 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/jvm/src/main/scala/fs2/io/net/NetworkPlatform.scala

@@ -25,6 +25,7 @@ package net
 
 import cats.effect.IO
 import cats.effect.LiftIO
+import cats.effect.Selector
 import cats.effect.kernel.{Async, Resource}
 
 import com.comcast.ip4s.{Dns, Host, IpAddress, Port, SocketAddress}
@@ -78,10 +79,62 @@ private[net] trait NetworkCompanionPlatform extends NetworkLowPriority { self: N
 
   def forIO: Network[IO] = forLiftIO
 
-  implicit def forLiftIO[F[_]: Async: LiftIO]: Network[F] = {
-    val _ = LiftIO[F]
-    forAsync
-  }
+  implicit def forLiftIO[F[_]: Async: LiftIO]: Network[F] =
+    new UnsealedNetwork[F] {
+      private lazy val fallback = forAsync[F]
+
+      private def tryGetSelector =
+        IO.pollers.map(_.collectFirst { case selector: Selector => selector }).to[F]
+
+      private implicit def dns: Dns[F] = Dns.forAsync[F]
+
+      def socketGroup(threadCount: Int, threadFactory: ThreadFactory): Resource[F, SocketGroup[F]] =
+        Resource.eval(tryGetSelector).flatMap {
+          case Some(selector) => Resource.pure(new SelectingSocketGroup[F](selector))
+          case None           => fallback.socketGroup(threadCount, threadFactory)
+        }
+
+      def datagramSocketGroup(threadFactory: ThreadFactory): Resource[F, DatagramSocketGroup[F]] =
+        fallback.datagramSocketGroup(threadFactory)
+
+      def client(
+          to: SocketAddress[Host],
+          options: List[SocketOption]
+      ): Resource[F, Socket[F]] = Resource.eval(tryGetSelector).flatMap {
+        case Some(selector) => new SelectingSocketGroup(selector).client(to, options)
+        case None           => fallback.client(to, options)
+      }
+
+      def server(
+          address: Option[Host],
+          port: Option[Port],
+          options: List[SocketOption]
+      ): Stream[F, Socket[F]] = Stream.eval(tryGetSelector).flatMap {
+        case Some(selector) => new SelectingSocketGroup(selector).server(address, port, options)
+        case None           => fallback.server(address, port, options)
+      }
+
+      def serverResource(
+          address: Option[Host],
+          port: Option[Port],
+          options: List[SocketOption]
+      ): Resource[F, (SocketAddress[IpAddress], Stream[F, Socket[F]])] =
+        Resource.eval(tryGetSelector).flatMap {
+          case Some(selector) =>
+            new SelectingSocketGroup(selector).serverResource(address, port, options)
+          case None => fallback.serverResource(address, port, options)
+        }
+
+      def openDatagramSocket(
+          address: Option[Host],
+          port: Option[Port],
+          options: List[SocketOption],
+          protocolFamily: Option[ProtocolFamily]
+      ): Resource[F, DatagramSocket[F]] =
+        fallback.openDatagramSocket(address, port, options, protocolFamily)
+
+      def tlsContext: TLSContext.Builder[F] = TLSContext.Builder.forAsync[F]
+    }
 
   def forAsync[F[_]](implicit F: Async[F]): Network[F] =
     forAsyncAndDns(F, Dns.forAsync(F))

io/jvm/src/main/scala/fs2/io/net/SelectingSocket.scala

@@ -0,0 +1,102 @@
+/*
+ * 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
+package io.net
+
+import cats.effect.LiftIO
+import cats.effect.Selector
+import cats.effect.kernel.Async
+import cats.effect.std.Mutex
+import cats.syntax.all._
+import com.comcast.ip4s.IpAddress
+import com.comcast.ip4s.SocketAddress
+
+import java.nio.ByteBuffer
+import java.nio.channels.SelectionKey.OP_READ
+import java.nio.channels.SelectionKey.OP_WRITE
+import java.nio.channels.SocketChannel
+
+private final class SelectingSocket[F[_]: LiftIO] private (
+    selector: Selector,
+    ch: SocketChannel,
+    readMutex: Mutex[F],
+    writeMutex: Mutex[F],
+    val localAddress: F[SocketAddress[IpAddress]],
+    val remoteAddress: F[SocketAddress[IpAddress]]
+)(implicit F: Async[F])
+    extends Socket.BufferedReads(readMutex) {
+
+  protected def readChunk(buf: ByteBuffer): F[Int] =
+    F.delay(ch.read(buf)).flatMap { readed =>
+      if (readed == 0) selector.select(ch, OP_READ).to *> readChunk(buf)
+      else F.pure(readed)
+    }
+
+  def write(bytes: Chunk[Byte]): F[Unit] = {
+    def go(buf: ByteBuffer): F[Unit] =
+      F.delay {
+        ch.write(buf)
+        buf.remaining()
+      }.flatMap { remaining =>
+        if (remaining > 0) {
+          selector.select(ch, OP_WRITE).to *> go(buf)
+        } else F.unit
+      }
+    writeMutex.lock.surround {
+      F.delay(bytes.toByteBuffer).flatMap(go)
+    }
+  }
+
+  def isOpen: F[Boolean] = F.delay(ch.isOpen)
+
+  def endOfOutput: F[Unit] =
+    F.delay {
+      ch.shutdownOutput(); ()
+    }
+
+  def endOfInput: F[Unit] =
+    F.delay {
+      ch.shutdownInput(); ()
+    }
+
+}
+
+private object SelectingSocket {
+  def apply[F[_]: LiftIO](
+      selector: Selector,
+      ch: SocketChannel,
+      localAddress: F[SocketAddress[IpAddress]],
+      remoteAddress: F[SocketAddress[IpAddress]]
+  )(implicit F: Async[F]): F[Socket[F]] =
+    (Mutex[F], Mutex[F]).flatMapN { (readMutex, writeMutex) =>
+      F.delay {
+        new SelectingSocket[F](
+          selector,
+          ch,
+          readMutex,
+          writeMutex,
+          localAddress,
+          remoteAddress
+        )
+      }
+    }
+}