Implement UringSystem and FS2 Sockets using netty io_uring API #78

antoniojimeneznieto · 2023-05-30T17:16:39Z

This pull request aims to adapt reusable code from Scala Native to Scala JVM. To provide compatibility with the uring library in JVM, we want to import functions from the Netty io_uring (https://github.com/netty/netty-incubator-transport-io_uring).

Plan:

Copy all the reusable code from Native to JVM
Import all the Netty functions for compatibility with the uring library
Reimplement the Native code in Scala JVM using Netty
- Implement UringSystem
- Implement UringSocket
- Implement UringSocketGroup
- Implement UringNetwork

antoniojimeneznieto · 2023-06-02T16:01:41Z

If I'm not mistaken, we are going to be using netty's Native class. However, it is not a public class. Do we make a fork of Netty and make it public in our version so we can use it ?

armanbilge · 2023-06-02T16:06:14Z

For now, we can create a "backdoor" by adding some code in the io.netty.incubator.channel.uring package that publicly exposes those methods so that we can use them.

antoniojimeneznieto · 2023-06-05T10:44:38Z

I wanted to see if I am on the right track 🤔. The next goal is to implement the UringSystem Poller, for that we will need to replace:
Ptr[io_uring] -> RingBuffer
Ptr[io_uring_sqe] -> IOUringSubmissionQueue
Ptr[Ptr[io_uring_sqe] -> List[IOUringSubmissionQueue]

armanbilge · 2023-06-05T11:20:50Z

That sounds right.

Ptr[Ptr[io_uring_sqe] -> List[IOUringSubmissionQueue]

Not sure about this one. Which method is this?

antoniojimeneznieto · 2023-06-05T12:21:13Z

In the processCqes:
def processCqes(_cqes: Ptr[Ptr[io_uring_cqe]])

armanbilge · 2023-06-05T12:26:12Z

Ah, you can access that one with the IOUringCompletionQueue.

SQE = submission queue entry (where you submit I/O work)
CQE = completion queue entry (where you collect results of completed work)

antoniojimeneznieto · 2023-06-08T13:02:13Z

Maybe I'm forgetting something but we don't seem to have access to FileDescriptorPoller and FileDescriptorPollHandle. What should we do 🤔 ?

armanbilge · 2023-06-08T15:34:32Z

Maybe I'm forgetting something but we don't seem to have access to FileDescriptorPoller and FileDescriptorPollHandle. What should we do 🤔 ?

These only make sense for Scala Native, you don't need to implement them for JVM :) you can delete it.

antoniojimeneznieto · 2023-06-08T16:23:01Z

Alright I see thanks :)

antoniojimeneznieto · 2023-06-08T17:52:25Z

I am looking for the equivalent to io_uring_prep_cancel64 for implementing the cancel method in the ApiImpl class and I am not completely sure, is it addPollRemove from IOUringSubmissionQueue 🤔 ?

armanbilge · 2023-06-08T17:57:46Z

No, that's something different, I wonder if Netty is missing it 🤔

But it doesn't matter, you can implement it manually using this method:

https://github.com/netty/netty-incubator-transport-io_uring/blob/c17c887b63bef590b83cc2f241a7ebbd16a9fd56/transport-classes-io_uring/src/main/java/io/netty/incubator/channel/uring/IOUringSubmissionQueue.java#L113-L114

Here is how liburing does it.
https://github.com/axboe/liburing/blob/b4ee3108b93f7e4602430246236d14978abad085/src/include/liburing.h#L632-L638

build.sbt

armanbilge

Ok, so getting serious about landing this ...

If you don't mind one last request: can we make an effort to deduplicate some of code and share between JVM/Native?

For implementation stuff, only if it's easy, I don't care too much (maybe for IOExceptionHelper, UringApp, implicits).

But ... I think we should try hard to share the tests, so we are running identical test suites on both platforms. We're not doing anything JVM-specific, hopefully?

armanbilge · 2023-09-19T15:42:18Z

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

+
+import java.net.ProtocolFamily
+
+private final class UringDatagramSocketGroup[F[_]] extends DatagramSocketGroup[F] {


I guess we can work on this in a follow-up PR :)

If you don't mind one last request: can we make an effort to deduplicate some of code and share between JVM/Native?

Great idea! I'll handle it :)

But ... I think we should try hard to share the tests, so we are running identical test suites on both platforms. We're not doing anything JVM-specific, hopefully?

I think we aren't, so it should be easy to do.

build.sbt

armanbilge · 2023-09-23T01:31:28Z

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

+          _ <- F.delay {
+            buffer.clear()
+            buffer.writeBytes(bytes.toArray)
+          }


Looks like we are re-using the same buffer for reading and writing? This is a problem if there are concurrent reads and writes, so we should use two buffers.

Also, I think we need logic to increase the buffer size for larger reads/writes?

Absolutely, my bad. Right now we would have two buffers with a predetermined size, we could create the buffer in the read/write itself depending on the number of bytes we want to send. Another option could be to have a default size and if the message is bigger than that size increase the buffer ?

Another option could be to have a default size and if the message is bigger than that size increase the buffer

Yes, this is how it's currently working in the other implementations: we keep replacing the buffer with a bigger one as-needed.

armanbilge · 2023-09-23T01:33:14Z

uring/jvm/src/main/scala/fs2/io/uring/unsafe/util.scala

+  def createBuffer[F[_]: Sync](size: Int): Resource[F, ByteBuf] =
+    Resource.make(
+      Sync[F].delay(UnpooledByteBufAllocator.DEFAULT.directBuffer(size))
+    )(buf => Sync[F].delay(if (buf.refCnt() > 0) { val _ = buf.release() }))


Hmm, what happens if buf.refCnt() == 0, who is responsible for cleaning it up?

I might be wrong but I think when the counter reaches 0 it is automatically deallocated by netty and if refCnt() returns 0 it has already been deallocated. However, I realized it should be:

buf.release(buf.refCnt()) to decrease the counter to 0 and be deallocated properly.

Hmm, I think that makes more sense.

Well I won't pretend to really understand this 😁 let's just make Netty go away sooner rather than later 😉

uring/jvm/src/main/scala/fs2/io/uring/unsafe/util.scala

armanbilge · 2023-09-23T01:55:26Z

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

+      linuxSocket => closeSocket(ring, linuxSocket.fd()).to
+    )
+
+  // private[this] def uringOpenSocket(ring: Uring, ipv6: Boolean): Resource[F, Int] = {


Can we use this?

Yes, the problem is that the UringSocket depends on the LinuxSocket to get the localAddress:

def localAddress: F[SocketAddress[IpAddress]] = F.delay(SocketAddress.fromInetSocketAddress(linuxSocket.getLocalAddress()))

When I implemented it I didn't find any API in netty to get the localAddress from the fd that UringOpenSocket gives us. I was planning to implement it once we remove netty but I'm going to check it, maybe this time I'll find something 🤞

Right, but we can make a LinuxSocket directly from the fd. Well, we need to hack the package 😅

https://github.com/netty/netty/blob/142027220ab762429c750c3089cfa31d7372dcd2/transport-classes-epoll/src/main/java/io/netty/channel/epoll/LinuxSocket.java#L48-L50

Good point! In that case it's done in a moment. Thanks Arman 😃

Co-authored-by: Arman Bilge <armanbilge@gmail.com>

antoniojimeneznieto added 3 commits May 30, 2023 18:45

Add unsafe to jvm

786abd1

Add uring files from native

16e6305

Import Netty io_uring

46f3dc8

Expose Native methods and RingBuffer

fd2c459

antoniojimeneznieto added 11 commits June 5, 2023 16:33

Add Completion and Submission Queues

be118ef

Add RingBuffer

6e9ea0a

Add header

88f02e7

Replace Ptr with SubmissionQueue

7be84f6

Call UringSystem as pollingSystem

08eb506

Add values for Completion and Submission queues

486a164

Start implementation of Poller

09b79cb

Add Companion Object

d767b7b

Sketch additional methods needed

408252c

Sketch reimplementation of UringSystem

3cb48c7

Add POLLIN & POLLOUT

d6e7d76

antoniojimeneznieto added 3 commits June 8, 2023 18:31

Remove FileDescriptor Poller and Handle

2b0f5aa

Formatting

0d1ae85

Formatting

18dd9f3

Add prepCancel

3fdda21

armanbilge added 2 commits August 28, 2023 13:07

Fixup

4e332b9

Fix netty deps

b655bb4

armanbilge mentioned this pull request Aug 28, 2023

Benchmark polling system powered Ember and friends typelevel/cats-effect#3692

Open

antoniojimeneznieto and others added 2 commits August 29, 2023 00:02

Add timeout test

dfbef6b

Fixups

51d90f8

armanbilge reviewed Aug 30, 2023

View reviewed changes

build.sbt Outdated Show resolved Hide resolved

antoniojimeneznieto added 4 commits September 1, 2023 12:56

Update netty version

d9b481b

Add warmup for timeout test

fba3875

Use send_msg for interruption and sketch getCurrentPollerIfAvailable

cea2b12

fix discarded non-Unit error

c8678d7

mox692 mentioned this pull request Sep 10, 2023

Add EpollSystem implementation for JVM armanbilge/fs2-epoll#2

Draft

armanbilge reviewed Sep 19, 2023

View reviewed changes

antoniojimeneznieto added 3 commits September 21, 2023 01:20

Remove datagramsocket

5f9922c

Share App, implicits and ExceptionHelper

7afdf02

Share UringSuite and TcpSocketSuite

b4cb557

armanbilge reviewed Sep 22, 2023

View reviewed changes

build.sbt Outdated Show resolved Hide resolved

build.sbt Outdated Show resolved Hide resolved

antoniojimeneznieto added 3 commits September 22, 2023 22:45

formatting

53f32de

undo shared in build

6342f80

formatting

3471231

armanbilge reviewed Sep 23, 2023

View reviewed changes

antoniojimeneznieto and others added 10 commits September 23, 2023 12:18

Update uring/jvm/src/main/scala/fs2/io/uring/unsafe/util.scala

a61f589

Co-authored-by: Arman Bilge <armanbilge@gmail.com>

Open sockets using uring

b8ee8d3

Fix buffer release bug

58e628f

Change masking error to negative

fc7f1a4

Use resizable buffer for reading and wriing

21b2af7

Replace buffer size with a constant value

b678106

Rename register to access

eee6522

Rename register to access

32a3282

Add license

b5cfd0b

Mock Cheshire API

6880b0c

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Implement UringSystem and FS2 Sockets using netty io_uring API #78

Implement UringSystem and FS2 Sockets using netty io_uring API #78

antoniojimeneznieto commented May 30, 2023 •

edited

Loading

antoniojimeneznieto commented Jun 2, 2023

armanbilge commented Jun 2, 2023

antoniojimeneznieto commented Jun 5, 2023 •

edited

Loading

armanbilge commented Jun 5, 2023

antoniojimeneznieto commented Jun 5, 2023 •

edited

Loading

armanbilge commented Jun 5, 2023

antoniojimeneznieto commented Jun 8, 2023

armanbilge commented Jun 8, 2023

antoniojimeneznieto commented Jun 8, 2023

antoniojimeneznieto commented Jun 8, 2023

armanbilge commented Jun 8, 2023

armanbilge left a comment

armanbilge Sep 19, 2023

antoniojimeneznieto Sep 19, 2023

armanbilge Sep 23, 2023

antoniojimeneznieto Sep 23, 2023

armanbilge Sep 23, 2023

armanbilge Sep 23, 2023

antoniojimeneznieto Sep 23, 2023

armanbilge Sep 23, 2023

armanbilge Sep 23, 2023

antoniojimeneznieto Sep 23, 2023

armanbilge Sep 23, 2023

antoniojimeneznieto Sep 23, 2023


		import java.net.ProtocolFamily

		private final class UringDatagramSocketGroup[F[_]] extends DatagramSocketGroup[F] {

Implement UringSystem and FS2 Sockets using netty io_uring API #78

Are you sure you want to change the base?

Implement UringSystem and FS2 Sockets using netty io_uring API #78

Conversation

antoniojimeneznieto commented May 30, 2023 • edited Loading

antoniojimeneznieto commented Jun 2, 2023

armanbilge commented Jun 2, 2023

antoniojimeneznieto commented Jun 5, 2023 • edited Loading

armanbilge commented Jun 5, 2023

antoniojimeneznieto commented Jun 5, 2023 • edited Loading

armanbilge commented Jun 5, 2023

antoniojimeneznieto commented Jun 8, 2023

armanbilge commented Jun 8, 2023

antoniojimeneznieto commented Jun 8, 2023

antoniojimeneznieto commented Jun 8, 2023

armanbilge commented Jun 8, 2023

armanbilge left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

antoniojimeneznieto commented May 30, 2023 •

edited

Loading

antoniojimeneznieto commented Jun 5, 2023 •

edited

Loading

antoniojimeneznieto commented Jun 5, 2023 •

edited

Loading