Implement UringSystem and FS2 Sockets using netty io_uring API

build.sbt

@@ -34,8 +34,14 @@ ThisBuild / githubWorkflowPublishPreamble +=
 
 val ceVersion = "3.6-e9aeb8c"
 val fs2Version = "3.8.0"
+val nettyVersion = "0.0.22.Final"
 val munitCEVersion = "2.0.0-M3"
 
+lazy val classifier = System.getProperty("os.arch") match {
+  case "amd64"   => "linux-x86_64"
+  case "aarch64" => "linux-aarch_64"
+}
+
 ThisBuild / nativeConfig ~= { c =>
   if (Option(System.getenv("CI")).contains("true"))
     c.withLinkingOptions(c.linkingOptions :+ "-luring")
@@ -57,3 +63,11 @@ lazy val uring = crossProject(NativePlatform, JVMPlatform)
     ),
     Test / testOptions += Tests.Argument("+l")
   )
+  .jvmSettings(
+    libraryDependencies ++= Seq(
+      "io.netty.incubator" % "netty-incubator-transport-classes-io_uring" % nettyVersion,
+      ("io.netty.incubator" % "netty-incubator-transport-native-io_uring" % nettyVersion % Test)
+        .classifier(classifier)
+    ),
+    fork := true
+  )

uring/jvm/src/main/scala/fs2/io/uring/IOExceptionHelper.scala

@@ -0,0 +1,76 @@
+/*
+ * Copyright 2022 Arman Bilge
+ *
+ * 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 fs2.io.uring
+
+import java.io.IOException
+import java.net.ConnectException
+import java.net.BindException
+import java.net.SocketException
+import java.net.SocketTimeoutException
+import java.net.NoRouteToHostException
+
+private[uring] object IOExceptionHelper {
+
+  def apply(errno: Int): IOException = errno match {
+    case 9 => // EBADF
+      new IOException("Bad file descriptor")
+
+    case 11 => // EAGAIN
+      new IOException("Resource temporarily unavailable")
+
+    case 13 => // EACCES
+      new IOException("Permission denied")
+
+    case 14 => // EFAULT
+      new IOException("Bad address")
+
+    case 22 => // EINVAL
+      new IOException("Invalid argument")
+
+    case 24 => // EMFILE
+      new IOException("Too many open files")
+
+    case 28 => // ENOSPC
+      new IOException("No space left on device")
+
+    case 32 => // EPIPE
+      new IOException("Broken pipe")
+
+    case 98 => // EADDRINUSE
+      new BindException("Address already in use")
+
+    case 99 => // EADDRNOTAVAIL
+      new BindException("Cannot assign requested address")
+
+    case 107 => // ECONNABORTED
+      new SocketException("Connection aborted")
+
+    case 110 => // ETIMEDOUT
+      new SocketTimeoutException("Connection timed out")
+
+    case 111 => // ECONNREFUSED
+      new ConnectException("Connection refused")
+
+    case 113 => // EHOSTUNREACH
+      new NoRouteToHostException("No route to host")
+
+    case 104 => // ECONNRESET
+      new SocketException("Connection reset by peer")
+
+    case _ => new IOException(errno.toString)
+  }
+}

uring/jvm/src/main/scala/fs2/io/uring/Uring.scala

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2022 Arman Bilge
+ *
+ * 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 fs2.io.uring
+
+import cats.effect.IO
+import cats.effect.LiftIO
+import cats.effect.kernel.Resource
+import cats.syntax.all._
+
+abstract class Uring private[uring] {
+  private[this] val noopMask: Int => Boolean = _ => false
+
+  def call(
+      op: Byte,
+      flags: Int = 0,
+      rwFlags: Int = 0,
+      fd: Int = 0,
+      bufferAddress: Long = 0,
+      length: Int = 0,
+      offset: Long = 0,
+      mask: Int => Boolean = noopMask
+  ): IO[Int]
+
+  def bracket(
+      op: Byte,
+      flags: Int = 0,
+      rwFlags: Int = 0,
+      fd: Int = 0,
+      bufferAddress: Long = 0,
+      length: Int = 0,
+      offset: Long = 0,
+      mask: Int => Boolean = noopMask
+  )(release: Int => IO[Unit]): Resource[IO, Int]
+}
+
+object Uring {
+
+  def get[F[_]: LiftIO]: F[Uring] =
+    IO.pollers.flatMap {
+      _.collectFirst { case ring: Uring =>
+        ring
+      }.liftTo[IO](new RuntimeException("No UringSystem installed"))
+    }.to
+
+}

uring/jvm/src/main/scala/fs2/io/uring/UringApp.scala

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2022 Arman Bilge
+ *
+ * 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 fs2.io.uring
+
+import cats.effect.IOApp
+import fs2.io.uring.unsafe.UringSystem
+
+trait UringApp extends IOApp {
+
+  override protected final def pollingSystem = UringSystem
+
+}
+
+object UringApp {
+  trait Simple extends IOApp.Simple with UringApp
+}

uring/jvm/src/main/scala/fs2/io/uring/implicits.scala

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2022 Arman Bilge
+ *
+ * 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 fs2.io.uring
+
+import cats.effect.LiftIO
+import cats.effect.kernel.Async
+import com.comcast.ip4s.Dns
+import fs2.io.net.Network
+
+object implicits {
+
+  @inline implicit def network[F[_]: Async: Dns: LiftIO]: Network[F] = ???
+
+}

uring/jvm/src/main/scala/fs2/io/uring/net/UringDatagramSocketGroup.scala

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2022 Arman Bilge
+ *
+ * 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 fs2.io.uring.net
+
+import cats.effect.kernel.Resource
+
+import com.comcast.ip4s._
+
+import fs2.io.net.DatagramSocketGroup
+import fs2.io.net.DatagramSocket
+import fs2.io.net._
+
+import java.net.ProtocolFamily
+
+private final class UringDatagramSocketGroup[F[_]] extends DatagramSocketGroup[F] {
+
+  override def openDatagramSocket(
+      address: Option[Host],
+      port: Option[Port],
+      options: List[DatagramSocketOption],
+      protocolFamily: Option[ProtocolFamily]
+  ): Resource[F, DatagramSocket[F]] = ???
+
+}
+
+object UringDatagramSocketGroup {
+  def apply[F[_]]: DatagramSocketGroup[F] = new UringDatagramSocketGroup
+}

uring/jvm/src/main/scala/fs2/io/uring/net/UringNetwork.scala

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2022 Arman Bilge
+ *
+ * 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 fs2.io.uring.net
+
+import cats.effect.LiftIO
+import cats.effect.kernel.Async
+import cats.effect.kernel.Resource
+
+import com.comcast.ip4s.Dns
+import com.comcast.ip4s.Host
+import com.comcast.ip4s.Port
+import com.comcast.ip4s.SocketAddress
+import com.comcast.ip4s.IpAddress
+
+import fs2.Stream
+import fs2.io.net.Network
+import fs2.io.net.SocketOption
+import fs2.io.net.tls.TLSContext
+import fs2.io.net.SocketGroup
+import fs2.io.net.Socket
+import fs2.io.net.DatagramSocket
+import fs2.io.net.DatagramSocketGroup
+import fs2.io.net.DatagramSocketOption
+
+import java.net.ProtocolFamily
+
+import java.util.concurrent.ThreadFactory
+
+private[net] final class UringNetwork[F[_]](
+    sg: UringSocketGroup[F],
+    dsg: UringDatagramSocketGroup[F],
+    val tlsContext: TLSContext.Builder[F]
+) extends Network.UnsealedNetwork[F] {
+
+  def socketGroup(threadCount: Int, threadFactory: ThreadFactory): Resource[F, SocketGroup[F]] =
+    Resource.pure[F, SocketGroup[F]](sg)
+
+  def datagramSocketGroup(threadFactory: ThreadFactory): Resource[F, DatagramSocketGroup[F]] =
+    Resource.pure[F, DatagramSocketGroup[F]](dsg)
+
+  def client(to: SocketAddress[Host], options: List[SocketOption]): Resource[F, Socket[F]] =
+    sg.client(to, options)
+
+  def server(
+      address: Option[Host],
+      port: Option[Port],
+      options: List[SocketOption]
+  ): Stream[F, Socket[F]] = sg.server(address, port, options)
+
+  def serverResource(
+      address: Option[Host],
+      port: Option[Port],
+      options: List[SocketOption]
+  ): Resource[F, (SocketAddress[IpAddress], Stream[F, Socket[F]])] =
+    sg.serverResource(address, port, options)
+
+  def openDatagramSocket(
+      address: Option[Host],
+      port: Option[Port],
+      options: List[DatagramSocketOption],
+      protocolFamily: Option[ProtocolFamily]
+  ): Resource[F, DatagramSocket[F]] =
+    dsg.openDatagramSocket(address, port, options, protocolFamily)
+}
+
+object UringNetwork {
+  def apply[F[_]: Async: Dns: LiftIO]: Network[F] =
+    new UringNetwork(
+      new UringSocketGroup[F],
+      new UringDatagramSocketGroup[F],
+      TLSContext.Builder.forAsync[F]
+    )
+}