Skip to content

Support Unix Domain Sockets with io_uring transport #6026

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
tsegismont wants to merge 1 commit into
base: master
Choose a base branch
from
+56 −32
Draft

Support Unix Domain Sockets with io_uring transport #6026

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .github/workflows/ci-matrix-5.x.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,9 @@ jobs:
- os: ubuntu-latest
jdk: 11
profile: '-PNativeEpoll+DomainSockets'
- os: ubuntu-latest
jdk: 11
profile: '-PNativeIoUring+DomainSockets'
- os: ubuntu-latest
jdk: 25
- os: windows-2022
Expand Down
23 changes: 23 additions & 0 deletions vertx-core/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -780,6 +780,29 @@
</properties>
</profile>

<profile>
<id>NativeIoUring+DomainSockets</id>
<properties>
<vertx.surefire.nettyTransport>io_uring</vertx.surefire.nettyTransport>
<vertx.surefire.useDomainSockets>true</vertx.surefire.useDomainSockets>
<vertx.surefire.useModulePath>false</vertx.surefire.useModulePath>
</properties>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<configuration>
<!-- TODO: enable all tests later when NativeIoUring profile is re-enabled in CI -->
<test>
io.vertx.tests.net.NetBandwidthLimitingTest,io.vertx.tests.net.NetTest#testListenDomainSocketAddressNative,HttpDomainSocketTest#testListenDomainSocketAddressNative,
</test>
</configuration>
</plugin>
</plugins>
</build>
</profile>

<profile>
<id>NativeKQueue</id>
<properties>
Expand Down
27 changes: 17 additions & 10 deletions vertx-core/src/main/java/io/vertx/core/impl/transports/IoUringTransport.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright (c) 2011-2024 Contributors to the Eclipse Foundation
* Copyright (c) 2011-2026 Contributors to the Eclipse Foundation
*
* This program and the accompanying materials are made available under the
* terms of the Eclipse Public License 2.0 which is available at
Expand All @@ -12,7 +12,10 @@

import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFactory;
import io.netty.channel.IoHandlerFactory;
import io.netty.channel.ServerChannel;
import io.netty.channel.socket.DatagramChannel;
import io.netty.channel.socket.InternetProtocolFamily;
import io.netty.channel.unix.DomainSocketAddress;
Expand Down Expand Up @@ -57,7 +60,7 @@ public IoUringTransport() {

@Override
public boolean supportsDomainSockets() {
return false;
return true;
}

@Override
Expand All @@ -68,7 +71,7 @@ public boolean supportFileRegion() {
@Override
public SocketAddress convert(io.vertx.core.net.SocketAddress address) {
if (address.isDomainSocket()) {
throw new IllegalArgumentException("Domain socket not supported by IOUring transport");
return new DomainSocketAddress(address.path());
}
return Transport.super.convert(address);
}
Expand Down Expand Up @@ -109,15 +112,15 @@ public ChannelFactory<? extends DatagramChannel> datagramChannelFactory() {
@Override
public ChannelFactory<? extends Channel> channelFactory(boolean domainSocket) {
if (domainSocket) {
throw new IllegalArgumentException();
return IoUringDomainSocketChannel::new;
}
return IoUringSocketChannel::new;
}

@Override
public ChannelFactory<? extends ServerChannel> serverChannelFactory(boolean domainSocket) {
if (domainSocket) {
throw new IllegalArgumentException();
return IoUringServerDomainSocketChannel::new;
}
return IoUringServerSocketChannel::new;
}
Expand All @@ -131,7 +134,9 @@ public void configure(DatagramChannel channel, DatagramSocketOptions options) {
@Override
public void configure(TcpConfig options, boolean domainSocket, ServerBootstrap bootstrap) {
if (domainSocket) {
throw new IllegalArgumentException();
// Domain sockets don't support TCP-specific options
Transport.super.configure(options, domainSocket, bootstrap);
return;
}
bootstrap.option(IoUringChannelOption.SO_REUSEPORT, options.isReusePort());
if (options.isTcpFastOpen()) {
Expand All @@ -140,20 +145,22 @@ public void configure(TcpConfig options, boolean domainSocket, ServerBootstrap b
bootstrap.childOption(IoUringChannelOption.TCP_USER_TIMEOUT, options.getTcpUserTimeout());
bootstrap.childOption(IoUringChannelOption.TCP_QUICKACK, options.isTcpQuickAck());
bootstrap.childOption(IoUringChannelOption.TCP_CORK, options.isTcpCork());
Transport.super.configure(options, false, bootstrap);
Transport.super.configure(options, domainSocket, bootstrap);
}

@Override
public void configure(TcpConfig options, boolean domainSocket, Bootstrap bootstrap) {
if (domainSocket) {
throw new IllegalArgumentException();
// Domain sockets don't support TCP-specific options
Transport.super.configure(options, domainSocket, bootstrap);
return;
}
if (options.isTcpFastOpen()) {
bootstrap.option(IoUringChannelOption.TCP_FASTOPEN_CONNECT, options.isTcpFastOpen());
}
bootstrap.option(IoUringChannelOption.TCP_USER_TIMEOUT, options.getTcpUserTimeout());
bootstrap.option(IoUringChannelOption.TCP_QUICKACK, options.isTcpQuickAck());
bootstrap.option(IoUringChannelOption.TCP_CORK, options.isTcpCork());
Transport.super.configure(options, false, bootstrap);
Transport.super.configure(options, domainSocket, bootstrap);
}
}
35 changes: 13 additions & 22 deletions vertx-core/src/test/java/io/vertx/tests/net/NetBandwidthLimitingTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,19 @@

package io.vertx.tests.net;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.DeploymentOptions;
import io.vertx.core.Future;
import io.vertx.core.Promise;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.net.*;
import io.vertx.test.core.TestUtils;
import io.vertx.test.core.VertxTestBase;
import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileOutputStream;
Expand All @@ -23,19 +36,6 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import io.vertx.core.net.*;
import io.vertx.core.transport.Transport;
import org.junit.*;
import org.junit.rules.TemporaryFolder;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.DeploymentOptions;
import io.vertx.core.Future;
import io.vertx.core.Promise;
import io.vertx.core.buffer.Buffer;
import io.vertx.test.core.TestUtils;
import io.vertx.test.core.VertxTestBase;

import static io.vertx.core.net.NetServerOptions.DEFAULT_PORT;

public class NetBandwidthLimitingTest extends VertxTestBase {
Expand Down Expand Up @@ -84,7 +84,6 @@ protected void tearDown() throws Exception {

@Test
public void sendBufferThrottled() {
Assume.assumeFalse(TRANSPORT == Transport.IO_URING);
long startTime = System.nanoTime();

Buffer expected = TestUtils.randomBuffer(64 * 1024 * 4);
Expand Down Expand Up @@ -117,7 +116,6 @@ public void sendBufferThrottled() {

@Test
public void sendFileIsThrottled() throws Exception {
Assume.assumeFalse(TRANSPORT == Transport.IO_URING);
long startTime = System.nanoTime();

File fDir = testFolder.newFolder();
Expand Down Expand Up @@ -153,7 +151,6 @@ public void sendFileIsThrottled() throws Exception {

@Test
public void dataUploadIsThrottled() {
Assume.assumeFalse(TRANSPORT == Transport.IO_URING);
long startTime = System.nanoTime();

Buffer expected = TestUtils.randomBuffer(64 * 1024 * 4);
Expand Down Expand Up @@ -185,7 +182,6 @@ public void dataUploadIsThrottled() {

@Test
public void fileUploadIsThrottled() throws Exception {
Assume.assumeFalse(TRANSPORT == Transport.IO_URING);
long startTime = System.nanoTime();

File fDir = testFolder.newFolder();
Expand Down Expand Up @@ -220,7 +216,6 @@ public void fileUploadIsThrottled() throws Exception {

@Test
public void testSendBufferIsTrafficShapedWithSharedServers() throws Exception {
Assume.assumeFalse(TRANSPORT == Transport.IO_URING);
Buffer expected = TestUtils.randomBuffer(64 * 1024 * 4);

int numEventLoops = 4; // We start a shared TCP server with 4 event-loops
Expand Down Expand Up @@ -264,8 +259,6 @@ public void start(Promise<Void> startPromise) {

@Test
public void testDynamicInboundRateUpdate() {
Assume.assumeFalse(TRANSPORT == Transport.IO_URING);

Buffer expected = TestUtils.randomBuffer(64 * 1024 * 4);
NetServer server = netServer();

Expand Down Expand Up @@ -299,7 +292,6 @@ public void testDynamicInboundRateUpdate() {

@Test
public void testDynamicOutboundRateUpdate() {
Assume.assumeFalse(TRANSPORT == Transport.IO_URING);
long startTime = System.nanoTime();

Buffer expected = TestUtils.randomBuffer(64 * 1024 * 4);
Expand Down Expand Up @@ -338,7 +330,6 @@ public void testDynamicOutboundRateUpdate() {

@Test(expected = IllegalStateException.class)
public void testRateUpdateWhenServerStartedWithoutTrafficShaping() throws Exception {
Assume.assumeFalse(TRANSPORT == Transport.IO_URING);
NetServerOptions options = new NetServerOptions().setHost(DEFAULT_HOST).setPort(DEFAULT_PORT);
NetServer testServer = netServer(options);

Expand Down
Loading