From 11882cdb131626f068626af9d8e3d8e261b7e2d3 Mon Sep 17 00:00:00 2001 From: "antonio.jimenez" Date: Sun, 19 Jul 2026 02:11:12 +0200 Subject: [PATCH 1/7] Add io_uring SQE preparation for Native sockets --- .../fs2/io/net/UringSocketOperations.scala | 156 ++++++++++++++++++ 1 file changed, 156 insertions(+) create mode 100644 io/native/src/main/scala/fs2/io/net/UringSocketOperations.scala diff --git a/io/native/src/main/scala/fs2/io/net/UringSocketOperations.scala b/io/native/src/main/scala/fs2/io/net/UringSocketOperations.scala new file mode 100644 index 0000000000..c712e52c23 --- /dev/null +++ b/io/native/src/main/scala/fs2/io/net/UringSocketOperations.scala @@ -0,0 +1,156 @@ +/* + * 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 +package net + +import fs2.io.internal.NativeUtil.errnoToThrowable + +import scala.scalanative.unsafe._ +import scala.scalanative.unsigned._ + +private[net] object UringSocketOperations { + + private final val IORING_OP_ACCEPT = 13 + private final val IORING_OP_CONNECT = 16 + private final val IORING_OP_CLOSE = 19 + private final val IORING_OP_SEND = 26 + private final val IORING_OP_RECV = 27 + private final val IORING_OP_SOCKET = 45 + + private type U8 = CUnsignedChar + private type U16 = CUnsignedShort + private type S32 = CInt + private type U32 = CUnsignedInt + private type U64 = CUnsignedLongLong + + type io_uring_sqe = CStruct10[ + U8, + U8, + U16, + S32, + U64, + U64, + U32, + U32, + U64, + CArray[U64, Nat._3] + ] + + def io_uring_prep_socket( + sqe: Ptr[io_uring_sqe], + domain: Int, + socketType: Int, + protocol: Int, + flags: Int + ): Unit = { + val typed = + io_uring_prep_rw(IORING_OP_SOCKET, sqe, domain, null, protocol.toUInt, socketType.toULong) + !typed.at8 = flags.toUInt + } + + def io_uring_prep_connect( + sqe: Ptr[io_uring_sqe], + fd: Int, + address: Ptr[?], + addressLength: CUnsignedInt + ): Unit = { + io_uring_prep_rw(IORING_OP_CONNECT, sqe, fd, address, 0.toUInt, addressLength.toULong) + () + } + + def io_uring_prep_accept( + sqe: Ptr[io_uring_sqe], + fd: Int, + address: Ptr[?], + addressLength: Ptr[CUnsignedInt], + flags: Int + ): Unit = { + val typed = io_uring_prep_rw( + IORING_OP_ACCEPT, + sqe, + fd, + address, + 0.toUInt, + pointerValue(addressLength) + ) + !typed.at8 = flags.toUInt + } + + def io_uring_prep_recv( + sqe: Ptr[io_uring_sqe], + fd: Int, + buffer: Ptr[Byte], + length: Int, + flags: Int + ): Unit = { + val typed = io_uring_prep_rw(IORING_OP_RECV, sqe, fd, buffer, length.toUInt, 0.toULong) + !typed.at8 = flags.toUInt + } + + def io_uring_prep_send( + sqe: Ptr[io_uring_sqe], + fd: Int, + buffer: Ptr[Byte], + length: Int, + flags: Int + ): Unit = { + val typed = io_uring_prep_rw(IORING_OP_SEND, sqe, fd, buffer, length.toUInt, 0.toULong) + !typed.at8 = flags.toUInt + } + + def io_uring_prep_close(sqe: Ptr[io_uring_sqe], fd: Int): Unit = { + io_uring_prep_rw(IORING_OP_CLOSE, sqe, fd, null, 0.toUInt, 0.toULong) + () + } + + def checkResult(result: Int): Int = + if (result < 0) throw errnoToThrowable(-result) + else result + + private def io_uring_prep_rw( + opcode: Int, + sqe: Ptr[io_uring_sqe], + fd: Int, + address: Ptr[?], + length: CUnsignedInt, + offset: U64 + ): Ptr[io_uring_sqe] = { + !sqe.at1 = opcode.toUByte + !sqe.at2 = 0.toUByte + !sqe.at3 = 0.toUShort + !sqe.at4 = fd + !sqe.at5 = offset + !sqe.at6 = pointerValue(address) + !sqe.at7 = length + !sqe.at8 = 0.toUInt + sqe._10(0) = 0.toULong + sqe._10(1) = 0.toULong + sqe._10(2) = 0.toULong + sqe + } + + private def pointerValue(pointer: Ptr[?]): U64 = + if (pointer eq null) 0.toULong + else pointer.toLong.toULong +} From 5ffa95f89482640dbb887aa23bde29e0ae0db0ec Mon Sep 17 00:00:00 2001 From: "antonio.jimenez" Date: Sun, 19 Jul 2026 02:27:57 +0200 Subject: [PATCH 2/7] Use CE uring local snapshot version --- build.sbt | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/build.sbt b/build.sbt index 95f7a3b0db..efc780095b 100644 --- a/build.sbt +++ b/build.sbt @@ -384,10 +384,17 @@ lazy val root = tlCrossRootProject benchmark ) +val catsEffectUringVersion = "3.7.0-uring-SNAPSHOT" + lazy val commonNativeSettings = Seq[Setting[?]]( tlVersionIntroduced := List("2.12", "2.13", "3").map(_ -> "3.13.0").toMap, Test / nativeBrewFormulas += "openssl", - Test / nativeConfig ~= { _.withEmbedResources(true) } + Test / nativeConfig ~= { _.withEmbedResources(true) }, + dependencyOverrides ++= Seq( + "org.typelevel" %%% "cats-effect" % catsEffectUringVersion, + "org.typelevel" %%% "cats-effect-kernel" % catsEffectUringVersion, + "org.typelevel" %%% "cats-effect-std" % catsEffectUringVersion + ) ) lazy val core = crossProject(JVMPlatform, JSPlatform, NativePlatform) From 9fdd2524a4edc1925c7f9478da3354c672d7e222 Mon Sep 17 00:00:00 2001 From: "antonio.jimenez" Date: Sun, 19 Jul 2026 02:50:31 +0200 Subject: [PATCH 3/7] Add UringSocket --- .../main/scala/fs2/io/net/UringSocket.scala | 162 ++++++++++++++++++ 1 file changed, 162 insertions(+) create mode 100644 io/native/src/main/scala/fs2/io/net/UringSocket.scala diff --git a/io/native/src/main/scala/fs2/io/net/UringSocket.scala b/io/native/src/main/scala/fs2/io/net/UringSocket.scala new file mode 100644 index 0000000000..8aaea34fde --- /dev/null +++ b/io/native/src/main/scala/fs2/io/net/UringSocket.scala @@ -0,0 +1,162 @@ +/* + * 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 +package net + +import cats.effect.unsafe.UringSystem.Uring +import cats.effect.{Async, LiftIO, Resource} +import cats.effect.std.Mutex +import cats.syntax.all._ +import com.comcast.ip4s.GenSocketAddress +import fs2.io.internal.NativeUtil.guardMask_ +import fs2.io.internal.{ResizableBuffer, SocketHelpers} + +import java.io.IOException +import scala.scalanative.posix.errno.ENOTCONN +import scala.scalanative.posix.sys.socket.{MSG_NOSIGNAL, shutdown} +import scala.scalanative.unsafe._ + +import UringSocketOperations._ + +private final class UringSocket[F[_]: LiftIO] private ( + ring: Uring, + fd: Int, + readBuffer: ResizableBuffer[F], + writeMutex: Mutex[F], + val address: GenSocketAddress, + val peerAddress: GenSocketAddress, + val isOpen: F[Boolean] +)(implicit F: Async[F]) + extends Socket[F] { + + def localAddress = F.pure(address.asIpUnsafe) + def remoteAddress = F.pure(peerAddress.asIpUnsafe) + + def endOfInput: F[Unit] = shutdownF(0) + def endOfOutput: F[Unit] = shutdownF(1) + private[this] def shutdownF(how: Int): F[Unit] = F.delay { + guardMask_(shutdown(fd, how))(_ == ENOTCONN) + } + + def read(maxBytes: Int): F[Option[Chunk[Byte]]] = + if (maxBytes < 0) + F.raiseError(new IllegalArgumentException("maxBytes must be non-negative")) + else if (maxBytes == 0) + F.pure(Some(Chunk.empty)) + else + readBuffer.get(maxBytes).use { buffer => + recv(buffer, maxBytes).flatMap { read => + if (read == 0) F.pure(None) + else F.delay(Some(Chunk.fromBytePtr(buffer, read))) + } + } + + def readN(numBytes: Int): F[Chunk[Byte]] = + if (numBytes < 0) + F.raiseError(new IllegalArgumentException("numBytes must be non-negative")) + else if (numBytes == 0) + F.pure(Chunk.empty) + else + readBuffer.get(numBytes).use { buffer => + def go(position: Int): F[Chunk[Byte]] = + recv(buffer + position.toLong, numBytes - position).flatMap { read => + val nextPosition = position + read + if (read == 0 || nextPosition == numBytes) + F.delay(Chunk.fromBytePtr(buffer, nextPosition)) + else + go(nextPosition) + } + + go(0) + } + + def reads: Stream[F, Byte] = + Stream.repeatEval(read(UringSocket.DefaultReadSize)).unNoneTerminate.unchunks + + def write(bytes: Chunk[Byte]): F[Unit] = + if (bytes.isEmpty) F.unit + else + writeMutex.lock.surround { + val Chunk.ArraySlice(buffer, offset, length) = bytes.toArraySlice + + def go(position: Int): F[Unit] = + send(buffer, offset + position, length - position).flatMap { sent => + val nextPosition = position + sent + if (sent == 0) + F.raiseError(new IOException("io_uring SEND completed without making progress")) + else if (nextPosition < length) go(nextPosition) + else F.unit + } + + go(0) + } + + def writes: Pipe[F, Byte, Nothing] = + _.chunks.foreach(write) + + def getOption[A](key: SocketOption.Key[A]) = + SocketHelpers.getOption[F, A](fd, key) + + def setOption[A](key: SocketOption.Key[A], value: A) = + SocketHelpers.setOption(fd, key, value) + + def supportedOptions = + SocketHelpers.supportedOptions + + ///////////////////////////////////////////////////////////////// + // io_uring helpers + ///////////////////////////////////////////////////////////////// + + private def recv(buffer: Ptr[Byte], length: Int): F[Int] = + ring + .call(io_uring_prep_recv(_, fd, buffer, length, 0), checkErrors = false) + .map(checkResult) + .to + + private def send(buffer: Array[Byte], offset: Int, length: Int): F[Int] = + ring + .call( + io_uring_prep_send(_, fd, buffer.atUnsafe(offset), length, MSG_NOSIGNAL), + checkErrors = false + ) + .map(result => checkResult(result) -> buffer) + .map(_._1) + .to +} + +private object UringSocket { + private final val DefaultReadSize = 8192 + + def apply[F[_]: LiftIO]( + ring: Uring, + fd: Int, + address: GenSocketAddress, + peerAddress: GenSocketAddress + )(implicit F: Async[F]): Resource[F, Socket[F]] = + for { + readBuffer <- ResizableBuffer(DefaultReadSize) + writeMutex <- Resource.eval(Mutex[F]) + isOpen <- Resource.make(F.ref(true))(_.set(false)) + } yield new UringSocket(ring, fd, readBuffer, writeMutex, address, peerAddress, isOpen.get) +} From 9c9eeb08d0022ae227d45cdaa983a70d4b068a9e Mon Sep 17 00:00:00 2001 From: "antonio.jimenez" Date: Sun, 19 Jul 2026 02:50:50 +0200 Subject: [PATCH 4/7] Add forUring Network --- .../main/scala/fs2/io/net/NetworkPlatform.scala | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/io/native/src/main/scala/fs2/io/net/NetworkPlatform.scala b/io/native/src/main/scala/fs2/io/net/NetworkPlatform.scala index af583a624e..82ee274e49 100644 --- a/io/native/src/main/scala/fs2/io/net/NetworkPlatform.scala +++ b/io/native/src/main/scala/fs2/io/net/NetworkPlatform.scala @@ -46,6 +46,21 @@ private[net] trait NetworkCompanionPlatform extends NetworkLowPriority { self: N ) } + def forUring[F[_]: Async: LiftIO]: Network[F] = + new AsyncProviderBasedNetwork[F] { + protected def mkIpSocketsProvider = + new UringIpSocketsProvider[F]()(Dns.forAsync, implicitly, implicitly) + protected def mkUnixSocketsProvider = new FdPollingUnixSocketsProvider[F] + protected def mkIpDatagramSocketsProvider = + throw new UnsupportedOperationException( + "Datagram sockets not currently supported on Native" + ) + protected def mkUnixDatagramSocketsProvider = + throw new UnsupportedOperationException( + "Unix datagram sockets not currently supported on Native" + ) + } + def forAsync[F[_]](implicit F: Async[F]): Network[F] = forAsyncAndDns(F, Dns.forAsync(F)) From e4d292ba4a211e352e51926c4117f654ee671e58 Mon Sep 17 00:00:00 2001 From: "antonio.jimenez" Date: Sun, 19 Jul 2026 02:51:17 +0200 Subject: [PATCH 5/7] Add UringIpSocketsProvider --- .../fs2/io/net/UringIpSocketsProvider.scala | 179 ++++++++++++++++++ 1 file changed, 179 insertions(+) create mode 100644 io/native/src/main/scala/fs2/io/net/UringIpSocketsProvider.scala diff --git a/io/native/src/main/scala/fs2/io/net/UringIpSocketsProvider.scala b/io/native/src/main/scala/fs2/io/net/UringIpSocketsProvider.scala new file mode 100644 index 0000000000..f7b0433360 --- /dev/null +++ b/io/native/src/main/scala/fs2/io/net/UringIpSocketsProvider.scala @@ -0,0 +1,179 @@ +/* + * 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 +package net + +import cats.effect.IO +import cats.effect.LiftIO +import cats.effect.kernel.Async +import cats.effect.kernel.Resource +import cats.effect.unsafe.UringSystem.Uring +import cats.syntax.all._ +import com.comcast.ip4s._ +import fs2.io.internal.NativeUtil._ +import fs2.io.internal.SocketHelpers +import fs2.io.internal.syssocket.{bind => sbind} + +import scala.scalanative.libc.stdlib +import scala.scalanative.posix.errno._ +import scala.scalanative.posix.netinet.in.sockaddr_in6 +import scala.scalanative.posix.sys.socket._ +import scala.scalanative.unsafe._ +import scala.scalanative.unsigned._ + +private final class UringIpSocketsProvider[F[_]: Dns: LiftIO](implicit F: Async[F]) + extends IpSocketsProvider[F] { + + override def connectIp( + to: SocketAddress[Host], + options: List[SocketOption] + ): Resource[F, Socket[F]] = for { + ring <- Resource.eval(Uring.get.to[F]) + address <- Resource.eval(to.resolve) + domain = domainFor(address.host) + fd <- openSocket(ring, domain) + _ <- Resource.eval(options.traverse_(so => SocketHelpers.setOption(fd, so.key, so.value))) + _ <- sockaddr(address).evalMap { case (addr, len) => + ring + .call(UringSocketOperations.io_uring_prep_connect(_, fd, addr, len), checkErrors = false) + .map(UringSocketOperations.checkResult) + .void + .to[F] + } + addresses <- Resource.eval(socketAddresses(fd, domain)) + socket <- UringSocket[F](ring, fd, addresses._1, addresses._2) + } yield socket + + override def bindIp( + address: SocketAddress[Host], + options: List[SocketOption] + ): Resource[F, ServerSocket[F]] = for { + ring <- Resource.eval(Uring.get.to[F]) + resolvedHost <- Resource.eval(address.host.resolve) + domain = domainFor(resolvedHost) + fd <- openSocket(ring, domain) + _ <- Resource.eval { + val bindF = F.delay { + val resolvedAddress = SocketAddress(resolvedHost, address.port) + SocketHelpers.toSockaddr(resolvedAddress) { (addr, len) => + guard_(sbind(fd, addr, len)) + } + } + + SocketHelpers.setOption(fd, SO_REUSEADDR, 1) *> + bindF *> + F.delay(guard_(listen(fd, 0))) + } + serverAddress <- Resource.eval(F.delay(SocketHelpers.getAddress(fd, domain))) + + sockets = Stream + .resource { + acceptSocket(ring, fd).flatMap { + case Some(clientFd) => + for { + _ <- Resource.eval( + options.traverse_(so => SocketHelpers.setOption(clientFd, so.key, so.value)) + ) + addresses <- Resource.eval(socketAddresses(clientFd, domain)) + socket <- UringSocket[F](ring, clientFd, addresses._1, addresses._2) + } yield Option(socket) + case None => Resource.eval(F.cede.as(Option.empty[Socket[F]])) + } + } + .repeat + .unNone + + info = new SocketInfo[F] { + def getOption[A](key: SocketOption.Key[A]) = SocketHelpers.getOption(fd, key) + def setOption[A](key: SocketOption.Key[A], value: A) = + SocketHelpers.setOption(fd, key, value) + def supportedOptions = SocketHelpers.supportedOptions + val address = serverAddress + } + } yield ServerSocket(info, sockets) + + private def openSocket(ring: Uring, domain: Int): Resource[F, Int] = + ring + .bracket( + UringSocketOperations.io_uring_prep_socket(_, domain, SOCK_STREAM, 0, 0), + checkErrors = false + )(closeSocket(ring, _)) + .mapK(LiftIO.liftK) + .evalMap(result => F.delay(UringSocketOperations.checkResult(result))) + + private def acceptSocket(ring: Uring, fd: Int): Resource[F, Option[Int]] = + ring + .bracket( + UringSocketOperations.io_uring_prep_accept(_, fd, null, null, 0), + checkErrors = false + )(closeSocket(ring, _)) + .mapK(LiftIO.liftK) + .evalMap { result => + F.delay { + if (retryAccept(result)) None + else Some(UringSocketOperations.checkResult(result)) + } + } + + private def retryAccept(result: Int): Boolean = { + val error = -result + result < 0 && + (error == EINTR || error == EAGAIN || error == EWOULDBLOCK || + error == ECONNABORTED || error == EPROTO || error == ENETDOWN || + error == ENOPROTOOPT || error == EHOSTUNREACH || error == EOPNOTSUPP || + error == ENETUNREACH) + } + + private def closeSocket(ring: Uring, fd: Int): IO[Unit] = + ring + .call(UringSocketOperations.io_uring_prep_close(_, fd), checkErrors = false) + .map(UringSocketOperations.checkResult) + .void + + private def socketAddresses(fd: Int, domain: Int) = + F.delay( + SocketHelpers.getAddress(fd, domain) -> SocketHelpers.getPeerAddress(fd, domain) + ) + + private def sockaddr( + address: SocketAddress[IpAddress] + ): Resource[F, (Ptr[sockaddr], socklen_t)] = + Resource.make { + F.delay { + val addr = + stdlib.calloc(1.toUSize, sizeof[sockaddr_in6]).asInstanceOf[Ptr[sockaddr]] + if (addr == null) + throw new OutOfMemoryError("Could not allocate sockaddr") + + val len = stackalloc[socklen_t]() + SocketHelpers.toSockaddr(address, addr, len) + (addr, !len) + } + } { case (addr, _) => + F.delay(stdlib.free(addr.asInstanceOf[Ptr[Byte]])) + } + + private def domainFor(address: IpAddress): Int = + if (address.isInstanceOf[Ipv4Address]) AF_INET else AF_INET6 +} From 359ad0093fbc57b9776b6d6ccfc247b96046aa04 Mon Sep 17 00:00:00 2001 From: "antonio.jimenez" Date: Sun, 19 Jul 2026 02:52:35 +0200 Subject: [PATCH 6/7] Make SocketSuite abstract and add implicit Network --- .../test/scala/fs2/io/net/SocketSuite.scala | 29 ++++++++++++++----- 1 file changed, 21 insertions(+), 8 deletions(-) diff --git a/io/shared/src/test/scala/fs2/io/net/SocketSuite.scala b/io/shared/src/test/scala/fs2/io/net/SocketSuite.scala index fe2e7c9e69..a4b1662546 100644 --- a/io/shared/src/test/scala/fs2/io/net/SocketSuite.scala +++ b/io/shared/src/test/scala/fs2/io/net/SocketSuite.scala @@ -30,7 +30,9 @@ import scala.concurrent.duration._ import scala.concurrent.TimeoutException import fs2.io.file._ -class SocketSuite extends Fs2Suite with SocketSuitePlatform { +abstract class SocketSuiteBase extends Fs2Suite with SocketSuitePlatform { + + implicit def network: Network[IO] val timeout = 30.seconds @@ -273,15 +275,22 @@ class SocketSuite extends Fs2Suite with SocketSuitePlatform { } test("sockets are released at the end of the resource scope") { - val f = - Network[IO].bind(SocketAddress.port(port"9071")).use { serverSocket => - serverSocket.accept.foreach(_ => IO.sleep(1.second)).compile.drain.background.surround { - Network[IO].connect(serverSocket.address).use { client => - client.read(1).assertEquals(None) - } + def exercise(serverSocket: ServerSocket[IO]) = + serverSocket.accept.foreach(_ => IO.sleep(1.second)).compile.drain.background.surround { + Network[IO].connect(serverSocket.address).use { client => + client.read(1).assertEquals(None) } } - f >> f >> f + + Network[IO] + .bind(SocketAddress(ip"127.0.0.1", Port.Wildcard)) + .use { serverSocket => + exercise(serverSocket).as(serverSocket.address) + } + .flatMap { address => + val f = Network[IO].bind(address).use(exercise) + f >> f + } } test("endOfOutput / endOfInput ignores ENOTCONN") { @@ -345,3 +354,7 @@ class SocketSuite extends Fs2Suite with SocketSuitePlatform { } } } + +class SocketSuite extends SocketSuiteBase { + override implicit lazy val network: Network[IO] = Network.forIO +} From b8403414b50ab522c78ffef6881dacbc2f581eeb Mon Sep 17 00:00:00 2001 From: "antonio.jimenez" Date: Sun, 19 Jul 2026 02:52:50 +0200 Subject: [PATCH 7/7] Add UringSocketSuite --- .../scala/fs2/io/net/UringSocketSuite.scala | 41 +++++++++++++++++++ 1 file changed, 41 insertions(+) create mode 100644 io/native/src/test/scala/fs2/io/net/UringSocketSuite.scala diff --git a/io/native/src/test/scala/fs2/io/net/UringSocketSuite.scala b/io/native/src/test/scala/fs2/io/net/UringSocketSuite.scala new file mode 100644 index 0000000000..a5633d4605 --- /dev/null +++ b/io/native/src/test/scala/fs2/io/net/UringSocketSuite.scala @@ -0,0 +1,41 @@ +/* + * 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.net + +import cats.effect.IO +import cats.effect.unsafe.{IORuntime, UringSystem} + +import scala.scalanative.meta.LinktimeInfo + +class UringSocketSuite extends SocketSuiteBase { + + override def munitIgnore: Boolean = !LinktimeInfo.isLinux + + override implicit lazy val network: Network[IO] = + if (LinktimeInfo.isLinux) Network.forUring[IO] + else Network.forIO + + override lazy val munitIORuntime: IORuntime = + if (LinktimeInfo.isLinux) IORuntime.builder().setPollingSystem(UringSystem).build() + else IORuntime.global +}