Merge pull request #3548 from onsah/issue/3182-jdk-unix-sockets

Use virtual threads when possible for JDK `UnixSocket`s

Author: armanbilge

io/jvm/src/main/scala/fs2/io/net/unixsocket/JdkUnixSockets.scala

@@ -24,6 +24,7 @@ package fs2.io.net.unixsocket
 import cats.effect.kernel.{Async, Resource}
 import cats.effect.syntax.all._
 import fs2.io.file.Files
+import fs2.io.evalOnVirtualThreadIfAvailable
 import java.net.{StandardProtocolFamily, UnixDomainSocketAddress}
 import java.nio.channels.{ServerSocketChannel, SocketChannel}
 
@@ -41,28 +42,32 @@ object JdkUnixSockets {
 private[unixsocket] class JdkUnixSocketsImpl[F[_]: Files](implicit F: Async[F])
     extends UnixSockets.AsyncUnixSockets[F] {
   protected def openChannel(address: UnixSocketAddress) =
-    Resource
-      .make(F.blocking(SocketChannel.open(StandardProtocolFamily.UNIX)))(ch =>
-        F.blocking(ch.close())
-      )
-      .evalTap { ch =>
-        F.blocking(ch.connect(UnixDomainSocketAddress.of(address.path)))
-          .cancelable(F.blocking(ch.close()))
-      }
+    evalOnVirtualThreadIfAvailable(
+      Resource
+        .make(
+          F.blocking(SocketChannel.open(StandardProtocolFamily.UNIX))
+        )(ch => F.blocking(ch.close()))
+        .evalTap { ch =>
+          F.blocking(ch.connect(UnixDomainSocketAddress.of(address.path)))
+            .cancelable(F.blocking(ch.close()))
+        }
+    )
 
   protected def openServerChannel(address: UnixSocketAddress) =
-    Resource
-      .make(F.blocking(ServerSocketChannel.open(StandardProtocolFamily.UNIX)))(ch =>
-        F.blocking(ch.close())
-      )
-      .evalTap { sch =>
-        F.blocking(sch.bind(UnixDomainSocketAddress.of(address.path)))
-          .cancelable(F.blocking(sch.close()))
-      }
-      .map { sch =>
-        Resource.makeFull[F, SocketChannel] { poll =>
-          poll(F.blocking(sch.accept).cancelable(F.blocking(sch.close())))
-        }(ch => F.blocking(ch.close()))
-      }
+    evalOnVirtualThreadIfAvailable(
+      Resource
+        .make(
+          F.blocking(ServerSocketChannel.open(StandardProtocolFamily.UNIX))
+        )(ch => F.blocking(ch.close()))
+        .evalTap { sch =>
+          F.blocking(sch.bind(UnixDomainSocketAddress.of(address.path)))
+            .cancelable(F.blocking(sch.close()))
+        }
+        .map { sch =>
+          Resource.makeFull[F, SocketChannel] { poll =>
+            poll(F.blocking(sch.accept).cancelable(F.blocking(sch.close())))
+          }(ch => F.blocking(ch.close()))
+        }
+    )
 
 }

io/jvm/src/main/scala/fs2/io/net/unixsocket/UnixSocketsPlatform.scala

@@ -31,6 +31,7 @@ import com.comcast.ip4s.{IpAddress, SocketAddress}
 import fs2.{Chunk, Stream}
 import fs2.io.file.{Files, Path}
 import fs2.io.net.Socket
+import fs2.io.evalOnVirtualThreadIfAvailable
 import java.nio.ByteBuffer
 import java.nio.channels.SocketChannel
 
@@ -101,15 +102,16 @@ private[unixsocket] trait UnixSocketsCompanionPlatform {
       extends Socket.BufferedReads[F](readMutex) {
 
     def readChunk(buff: ByteBuffer): F[Int] =
-      F.blocking(ch.read(buff)).cancelable(close)
+      evalOnVirtualThreadIfAvailable(F.blocking(ch.read(buff)))
+        .cancelable(close)
 
     def write(bytes: Chunk[Byte]): F[Unit] = {
       def go(buff: ByteBuffer): F[Unit] =
         F.blocking(ch.write(buff)).cancelable(close) *>
           F.delay(buff.remaining <= 0).ifM(F.unit, go(buff))
 
       writeMutex.lock.surround {
-        F.delay(bytes.toByteBuffer).flatMap(go)
+        F.delay(bytes.toByteBuffer).flatMap(buffer => evalOnVirtualThreadIfAvailable(go(buffer)))
       }
     }
 
@@ -118,15 +120,19 @@ private[unixsocket] trait UnixSocketsCompanionPlatform {
     private def raiseIpAddressError[A]: F[A] =
       F.raiseError(new UnsupportedOperationException("UnixSockets do not use IP addressing"))
 
-    def isOpen: F[Boolean] = F.blocking(ch.isOpen())
-    def close: F[Unit] = F.blocking(ch.close())
+    def isOpen: F[Boolean] = evalOnVirtualThreadIfAvailable(F.blocking(ch.isOpen()))
+    def close: F[Unit] = evalOnVirtualThreadIfAvailable(F.blocking(ch.close()))
     def endOfOutput: F[Unit] =
-      F.blocking {
-        ch.shutdownOutput(); ()
-      }
+      evalOnVirtualThreadIfAvailable(
+        F.blocking {
+          ch.shutdownOutput(); ()
+        }
+      )
     def endOfInput: F[Unit] =
-      F.blocking {
-        ch.shutdownInput(); ()
-      }
+      evalOnVirtualThreadIfAvailable(
+        F.blocking {
+          ch.shutdownInput(); ()
+        }
+      )
   }
 }