Implement UringSystem and FS2 Sockets using netty io_uring API by antoniojimeneznieto · Pull Request #78 · armanbilge/fs2-io_uring

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

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

+
+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.

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

+          _ <- 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

+  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 😉

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

+      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>

Adapts the Native UringSystem to the multi-poller runtime that Scala Native 0.5+ enables. Mirrors the structure of the JVM prototype from GSoC (armanbilge/fs2-io_uring#78)

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

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

Comment thread build.sbt Outdated

Comment thread build.sbt Outdated

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 18 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

Remove Cheshire as it was moved to a separete PR

6e3d1b3

Update Cats Effect version to 3.6.0-RC2

742ffda

Sketch UringSystem APIs using the new PollingSystem APIs

8b2dc44

Sketch PollerMetricsImpl, ApiImpl, PollerImpl

22aaa76

Sketch UringSystem for native

f9ee424

Remove PollerMetricsImpl and use PollingContext

00b89c9

Update Native UringSystem to use CE 3.6.0-RC2

c3ab8e3

WIP Update JVM UringSystem to use CE 3.6.0-RC2 use

d15a511


		import java.net.ProtocolFamily

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

Conversation

antoniojimeneznieto commented May 30, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

antoniojimeneznieto commented Jun 2, 2023

Uh oh!

armanbilge commented Jun 2, 2023

Uh oh!

antoniojimeneznieto commented Jun 5, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

armanbilge commented Jun 5, 2023

Uh oh!

antoniojimeneznieto commented Jun 5, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

armanbilge commented Jun 5, 2023

Uh oh!

antoniojimeneznieto commented Jun 8, 2023

Uh oh!

armanbilge commented Jun 8, 2023

Uh oh!

antoniojimeneznieto commented Jun 8, 2023

Uh oh!

antoniojimeneznieto commented Jun 8, 2023

Uh oh!

armanbilge commented Jun 8, 2023

Uh oh!

armanbilge left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

antoniojimeneznieto commented May 30, 2023 •

edited

Loading

antoniojimeneznieto commented Jun 5, 2023 •

edited

Loading

antoniojimeneznieto commented Jun 5, 2023 •

edited

Loading