Skip to content

Commit

Permalink
Allow domain sockets to configure SO_SNDBUF and SO_RCVBUF (#9584)
Browse files Browse the repository at this point in the history 
Motivation:

Running tests with a `KQueueDomainSocketChannel` showed worse performance than an `NioSocketChannel`. It turns out that the default send buffer size for Nio sockets is 64k while for KQueue sockets it's 8k. I verified that manually setting the socket's send buffer size improved perf to expected levels.

Modification:

Plumb the `SO_SNDBUF` and `SO_RCVBUF` options into the `*DomainSocketChannelConfig`.

Result:

Can now configure send and receive buffer sizes for domain sockets.
  • Loading branch information
@ellisjoe @normanmaurer
ellisjoe authored and normanmaurer committed Sep 20, 2019
1 parent 07fe1a2 commit aebe206
Show file tree
Hide file tree
Showing 2 changed files with 98 additions and 3 deletions.
49 changes: 48 additions & 1 deletion ...ort-native-epoll/src/main/java/io/netty/channel/epoll/EpollDomainSocketChannelConfig.java
View file
Expand Up @@ -24,9 +24,12 @@
import io.netty.channel.unix.DomainSocketChannelConfig;
import io.netty.channel.unix.DomainSocketReadMode;

import java.io.IOException;
import java.util.Map;

import static io.netty.channel.ChannelOption.ALLOW_HALF_CLOSURE;
import static io.netty.channel.ChannelOption.SO_RCVBUF;
import static io.netty.channel.ChannelOption.SO_SNDBUF;
import static io.netty.channel.unix.UnixChannelOption.DOMAIN_SOCKET_READ_MODE;

public final class EpollDomainSocketChannelConfig extends EpollChannelConfig
Expand All @@ -40,7 +43,7 @@ public final class EpollDomainSocketChannelConfig extends EpollChannelConfig

@Override
public Map<ChannelOption<?>, Object> getOptions() {
return getOptions(super.getOptions(), DOMAIN_SOCKET_READ_MODE, ALLOW_HALF_CLOSURE);
return getOptions(super.getOptions(), DOMAIN_SOCKET_READ_MODE, ALLOW_HALF_CLOSURE, SO_SNDBUF, SO_RCVBUF);
}

@SuppressWarnings("unchecked")
Expand All @@ -52,6 +55,12 @@ public <T> T getOption(ChannelOption<T> option) {
if (option == ALLOW_HALF_CLOSURE) {
return (T) Boolean.valueOf(isAllowHalfClosure());
}
if (option == SO_SNDBUF) {
return (T) Integer.valueOf(getSendBufferSize());
}
if (option == SO_RCVBUF) {
return (T) Integer.valueOf(getReceiveBufferSize());
}
return super.getOption(option);
}

Expand All @@ -63,6 +72,10 @@ public <T> boolean setOption(ChannelOption<T> option, T value) {
setReadMode((DomainSocketReadMode) value);
} else if (option == ALLOW_HALF_CLOSURE) {
setAllowHalfClosure((Boolean) value);
} else if (option == SO_SNDBUF) {
setSendBufferSize((Integer) value);
} else if (option == SO_RCVBUF) {
setReceiveBufferSize((Integer) value);
} else {
return super.setOption(option, value);
}
Expand Down Expand Up @@ -173,4 +186,38 @@ public EpollDomainSocketChannelConfig setAllowHalfClosure(boolean allowHalfClosu
this.allowHalfClosure = allowHalfClosure;
return this;
}

public int getSendBufferSize() {
try {
return ((EpollDomainSocketChannel) channel).socket.getSendBufferSize();
} catch (IOException e) {
throw new RuntimeException(e);
}
}

public EpollDomainSocketChannelConfig setSendBufferSize(int sendBufferSize) {
try {
((EpollDomainSocketChannel) channel).socket.setSendBufferSize(sendBufferSize);
return this;
} catch (IOException e) {
throw new RuntimeException(e);
}
}

public int getReceiveBufferSize() {
try {
return ((EpollDomainSocketChannel) channel).socket.getReceiveBufferSize();
} catch (IOException e) {
throw new RuntimeException(e);
}
}

public EpollDomainSocketChannelConfig setReceiveBufferSize(int receiveBufferSize) {
try {
((EpollDomainSocketChannel) channel).socket.setReceiveBufferSize(receiveBufferSize);
return this;
} catch (IOException e) {
throw new RuntimeException(e);
}
}
}
52 changes: 50 additions & 2 deletions ...-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueDomainSocketChannelConfig.java
View file
Expand Up @@ -25,9 +25,13 @@
import io.netty.channel.unix.DomainSocketReadMode;
import io.netty.util.internal.UnstableApi;

import java.io.IOException;
import java.util.Map;

import static io.netty.channel.unix.UnixChannelOption.*;
import static io.netty.channel.ChannelOption.ALLOW_HALF_CLOSURE;
import static io.netty.channel.ChannelOption.SO_RCVBUF;
import static io.netty.channel.ChannelOption.SO_SNDBUF;
import static io.netty.channel.unix.UnixChannelOption.DOMAIN_SOCKET_READ_MODE;

@UnstableApi
public final class KQueueDomainSocketChannelConfig extends KQueueChannelConfig implements DomainSocketChannelConfig {
Expand All @@ -40,7 +44,7 @@ public final class KQueueDomainSocketChannelConfig extends KQueueChannelConfig i

@Override
public Map<ChannelOption<?>, Object> getOptions() {
return getOptions(super.getOptions(), DOMAIN_SOCKET_READ_MODE, ALLOW_HALF_CLOSURE);
return getOptions(super.getOptions(), DOMAIN_SOCKET_READ_MODE, ALLOW_HALF_CLOSURE, SO_SNDBUF, SO_RCVBUF);
}

@SuppressWarnings("unchecked")
Expand All @@ -52,6 +56,12 @@ public <T> T getOption(ChannelOption<T> option) {
if (option == ALLOW_HALF_CLOSURE) {
return (T) Boolean.valueOf(isAllowHalfClosure());
}
if (option == SO_SNDBUF) {
return (T) Integer.valueOf(getSendBufferSize());
}
if (option == SO_RCVBUF) {
return (T) Integer.valueOf(getReceiveBufferSize());
}
return super.getOption(option);
}

Expand All @@ -63,6 +73,10 @@ public <T> boolean setOption(ChannelOption<T> option, T value) {
setReadMode((DomainSocketReadMode) value);
} else if (option == ALLOW_HALF_CLOSURE) {
setAllowHalfClosure((Boolean) value);
} else if (option == SO_SNDBUF) {
setSendBufferSize((Integer) value);
} else if (option == SO_RCVBUF) {
setReceiveBufferSize((Integer) value);
} else {
return super.setOption(option, value);
}
Expand Down Expand Up @@ -159,6 +173,40 @@ public DomainSocketReadMode getReadMode() {
return mode;
}

public int getSendBufferSize() {
try {
return ((KQueueDomainSocketChannel) channel).socket.getSendBufferSize();
} catch (IOException e) {
throw new RuntimeException(e);
}
}

public KQueueDomainSocketChannelConfig setSendBufferSize(int sendBufferSize) {
try {
((KQueueDomainSocketChannel) channel).socket.setSendBufferSize(sendBufferSize);
return this;
} catch (IOException e) {
throw new RuntimeException(e);
}
}

public int getReceiveBufferSize() {
try {
return ((KQueueDomainSocketChannel) channel).socket.getReceiveBufferSize();
} catch (IOException e) {
throw new RuntimeException(e);
}
}

public KQueueDomainSocketChannelConfig setReceiveBufferSize(int receiveBufferSize) {
try {
((KQueueDomainSocketChannel) channel).socket.setReceiveBufferSize(receiveBufferSize);
return this;
} catch (IOException e) {
throw new RuntimeException(e);
}
}

/**
* @see SocketChannelConfig#isAllowHalfClosure()
*/
Expand Down

0 comments on commit aebe206

Please sign in to comment.