andreacavalli
/
netty5
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Fix checkstyle errors

This commit is contained in:
Norman Maurer 2020-09-07 10:29:41 +02:00
parent 8c465e2f1b
commit ddb503f76d
11 changed files with 120 additions and 94 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
transport-native-io_uring/src/main/java/io/netty/channel/uring/AbstractIOUringChannel.java
View File

@ -34,7 +34,13 @@ import io.netty.channel.RecvByteBufAllocator;
import io.netty.channel.socket.ChannelInputShutdownEvent;
import io.netty.channel.socket.ChannelInputShutdownReadComplete;
import io.netty.channel.socket.SocketChannelConfig;
import io.netty.channel.unix.*;
import io.netty.channel.unix.Buffer;
import io.netty.channel.unix.Errors;
import io.netty.channel.unix.FileDescriptor;
import io.netty.channel.unix.NativeInetAddress;
import io.netty.channel.unix.Socket;
import io.netty.channel.unix.UnixChannel;
import io.netty.channel.unix.UnixChannelUtil;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;
@ -320,7 +326,6 @@ abstract class AbstractIOUringChannel extends AbstractChannel implements UnixCha
}
}
protected final void doWriteSingle(ByteBuf buf) {
IOUringSubmissionQueue submissionQueue = submissionQueue();
submissionQueue.addWrite(socket.intValue(), buf.memoryAddress(), buf.readerIndex(),
@ -520,7 +525,8 @@ abstract class AbstractIOUringChannel extends AbstractChannel implements UnixCha
fulfillConnectPromise(connectPromise, annotateConnectException(t, requestedRemoteAddress));
} finally {
if (!connectStillInProgress) {
// Check for null as the connectTimeoutFuture is only created if a connectTimeoutMillis > 0 is used
// Check for null as the connectTimeoutFuture is only created if a connectTimeoutMillis > 0
// is used
// See https://github.com/netty/netty/issues/1770
cancelConnectTimeoutFuture();
connectPromise = null;
@ -573,7 +579,8 @@ abstract class AbstractIOUringChannel extends AbstractChannel implements UnixCha
} catch (Throwable cause) {
fulfillConnectPromise(connectPromise, cause);
} finally {
// Check for null as the connectTimeoutFuture is only created if a connectTimeoutMillis > 0 is used
// Check for null as the connectTimeoutFuture is only created if a connectTimeoutMillis > 0 is
// used
// See https://github.com/netty/netty/issues/1770
cancelConnectTimeoutFuture();
connectPromise = null;
@ -601,7 +608,8 @@ abstract class AbstractIOUringChannel extends AbstractChannel implements UnixCha
remoteAddressMemory = Buffer.allocateDirectWithNativeOrder(SOCK_ADDR_LEN);
long remoteAddressMemoryAddress = Buffer.memoryAddress(remoteAddressMemory);
socket.initAddress(address.address(), address.scopeId(), inetSocketAddress.getPort(), remoteAddressMemoryAddress);
socket.initAddress(address.address(), address.scopeId(), inetSocketAddress.getPort(),
remoteAddressMemoryAddress);
final IOUringSubmissionQueue ioUringSubmissionQueue = submissionQueue();
ioUringSubmissionQueue.addConnect(socket.intValue(), remoteAddressMemoryAddress, SOCK_ADDR_LEN);
} catch (Throwable t) {
@ -629,7 +637,7 @@ abstract class AbstractIOUringChannel extends AbstractChannel implements UnixCha
promise.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
public void operationComplete(ChannelFuture future) {
if (future.isCancelled()) {
cancelConnectTimeoutFuture();
connectPromise = null;
@ -688,7 +696,7 @@ abstract class AbstractIOUringChannel extends AbstractChannel implements UnixCha
return remote;
}
public Socket getSocket() {
protected Socket getSocket() {
return socket;
}

28
transport-native-io_uring/src/main/java/io/netty/channel/uring/IOUring.java
View File

@ -15,6 +15,7 @@
*/
package io.netty.channel.uring;
import io.netty.util.internal.PlatformDependent;
import io.netty.util.internal.SystemPropertyUtil;
final class IOUring {
@ -28,19 +29,24 @@ final class IOUring {
cause = new UnsupportedOperationException(
"Native transport was explicit disabled with -Dio.netty.transport.noNative=true");
} else {
RingBuffer ringBuffer = null;
try {
ringBuffer = Native.createRingBuffer();
} catch (Throwable t) {
cause = t;
} finally {
if (ringBuffer != null) {
try {
ringBuffer.close();
} catch (Exception ignore) {
// ignore
Throwable unsafeCause = PlatformDependent.getUnsafeUnavailabilityCause();
if (unsafeCause == null) {
RingBuffer ringBuffer = null;
try {
ringBuffer = Native.createRingBuffer();
} catch (Throwable t) {
cause = t;
} finally {
if (ringBuffer != null) {
try {
ringBuffer.close();
} catch (Exception ignore) {
// ignore
}
}
}
} else {
cause = new UnsupportedOperationException("Unsafe is not supported", unsafeCause);
}
}

31
transport-native-io_uring/src/main/java/io/netty/channel/uring/IOUringServerSocketChannelConfig.java
View File

@ -16,16 +16,18 @@
package io.netty.channel.uring;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.ChannelException;
import io.netty.channel.ChannelOption;
import io.netty.channel.DefaultChannelConfig;
import io.netty.channel.MessageSizeEstimator;
import io.netty.channel.RecvByteBufAllocator;
import io.netty.channel.WriteBufferWaterMark;
import io.netty.channel.socket.ServerSocketChannelConfig;
import io.netty.util.NetUtil;
import java.io.IOException;
import java.util.Map;
import static io.netty.channel.ChannelOption.SO_BACKLOG;
import static io.netty.channel.ChannelOption.SO_RCVBUF;
import static io.netty.channel.ChannelOption.SO_REUSEADDR;
import static io.netty.util.internal.ObjectUtil.checkPositiveOrZero;
public final class IOUringServerSocketChannelConfig extends DefaultChannelConfig implements ServerSocketChannelConfig {
@ -38,21 +40,21 @@ public final class IOUringServerSocketChannelConfig extends DefaultChannelConfig
@Override
public Map<ChannelOption<?>, Object> getOptions() {
return getOptions(super.getOptions(), SO_RCVBUF, SO_REUSEADDR, SO_BACKLOG,
IOUringChannelOption.SO_REUSEPORT, IOUringChannelOption.IP_FREEBIND,
IOUringChannelOption.IP_TRANSPARENT, IOUringChannelOption.TCP_DEFER_ACCEPT);
return getOptions(super.getOptions(), ChannelOption.SO_RCVBUF, ChannelOption.SO_REUSEADDR,
ChannelOption.SO_BACKLOG, IOUringChannelOption.SO_REUSEPORT, IOUringChannelOption.IP_FREEBIND,
IOUringChannelOption.IP_TRANSPARENT, IOUringChannelOption.TCP_DEFER_ACCEPT);
}
@SuppressWarnings("unchecked")
@Override
public <T> T getOption(ChannelOption<T> option) {
if (option == SO_RCVBUF) {
if (option == ChannelOption.SO_RCVBUF) {
return (T) Integer.valueOf(getReceiveBufferSize());
}
if (option == SO_REUSEADDR) {
if (option == ChannelOption.SO_REUSEADDR) {
return (T) Boolean.valueOf(isReuseAddress());
}
if (option == SO_BACKLOG) {
if (option == ChannelOption.SO_BACKLOG) {
return (T) Integer.valueOf(getBacklog());
}
if (option == IOUringChannelOption.SO_REUSEPORT) {
@ -73,11 +75,11 @@ public final class IOUringServerSocketChannelConfig extends DefaultChannelConfig
@Override
public <T> boolean setOption(ChannelOption<T> option, T value) {
validate(option, value);
if (option == SO_RCVBUF) {
if (option == ChannelOption.SO_RCVBUF) {
setReceiveBufferSize((Integer) value);
} else if (option == SO_REUSEADDR) {
} else if (option == ChannelOption.SO_REUSEADDR) {
setReuseAddress((Boolean) value);
} else if (option == SO_BACKLOG) {
} else if (option == ChannelOption.SO_BACKLOG) {
setBacklog((Integer) value);
} else if (option == IOUringChannelOption.SO_REUSEPORT) {
setReusePort((Boolean) value);
@ -225,7 +227,8 @@ public final class IOUringServerSocketChannelConfig extends DefaultChannelConfig
/**
* Set the SO_REUSEPORT option on the underlying Channel. This will allow to bind multiple
* {@link io.netty.channel.socket.ServerSocketChannel}s to the same port and so accept connections with multiple threads.
* {@link io.netty.channel.socket.ServerSocketChannel}s to the same port and so accept connections with multiple
* threads.
*
* Be aware this method needs be called before
* {@link io.netty.channel.socket.ServerSocketChannel#bind(java.net.SocketAddress)} to have any affect.

2
transport-native-io_uring/src/main/java/io/netty/channel/uring/IOUringSubmissionQueue.java
View File

@ -357,7 +357,7 @@ final class IOUringSubmissionQueue {
}
public long count() {
return (sqeTail - toUnsignedLong(PlatformDependent.getIntVolatile(kHeadAddress)));
return sqeTail - toUnsignedLong(PlatformDependent.getIntVolatile(kHeadAddress));
}
//delete memory

27
transport-native-io_uring/src/main/java/io/netty/channel/uring/IovecArrayPool.java
View File

@ -1,3 +1,18 @@
/*
* Copyright 2020 The Netty Project
*
* The Netty Project licenses this file to you 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 io.netty.channel.uring;
import io.netty.buffer.ByteBuf;
@ -34,7 +49,7 @@ final class IovecArrayPool implements MessageProcessor {
private final ByteBuffer iovecArrayMemory;
private final long iovecArrayMemoryAddress;
public IovecArrayPool() {
IovecArrayPool() {
//setup array
remainingIovec = new Stack<Long>();
@ -47,8 +62,7 @@ final class IovecArrayPool implements MessageProcessor {
}
//Todo better naming
public long createNewIovecMemoryAddress() {
long createNewIovecMemoryAddress() {
//clear
size = 0;
count = 0;
@ -65,13 +79,12 @@ final class IovecArrayPool implements MessageProcessor {
}
//Todo error handling
public void releaseIovec(long iovecAddress) {
void releaseIovec(long iovecAddress) {
long index = (iovecAddress - iovecArrayMemoryAddress) / IOVEC_ARRAY_SIZE;
remainingIovec.push(index);
}
private boolean add(ByteBuf buf, int offset, int len) {
if (count == IOV_ENTRIES) {
// No more room!
@ -138,7 +151,7 @@ final class IovecArrayPool implements MessageProcessor {
return false;
}
public int count() {
int count() {
return count;
}
@ -146,7 +159,7 @@ final class IovecArrayPool implements MessageProcessor {
return IOV_SIZE * index;
}
public void release() {
void release() {
Buffer.free(iovecArrayMemory);
}
}

3
transport-native-io_uring/src/main/java/io/netty/channel/uring/LinuxSocket.java
View File

@ -387,5 +387,6 @@ final class LinuxSocket extends Socket {
private static native int getIpMulticastLoop(int fd, boolean ipv6) throws IOException;
private static native void setIpMulticastLoop(int fd, boolean ipv6, int enabled) throws IOException;
private static native void setTimeToLive(int fd, int ttl) throws IOException;
private static native int initAddress(int fd, boolean ipv6, byte[] address, int scopeId, int port, long memoryAddress);
private static native int initAddress(
int fd, boolean ipv6, byte[] address, int scopeId, int port, long memoryAddress);
}

5
transport-native-io_uring/src/test/java/io/netty/channel/uring/IOUringEventLoopTest.java
View File

@ -79,13 +79,10 @@ public class IOUringEventLoopTest extends AbstractSingleThreadEventLoopTest {
EventLoop loop = group.next();
loop.schedule(new Runnable() {
@Override
public void run() {
}
public void run() { }
}, 1, TimeUnit.SECONDS).sync();
} finally {
group.shutdownGracefully();
}
}
}

2
transport-native-io_uring/src/test/java/io/netty/channel/uring/IOUringSocketDataReadInitialStateTest.java
View File

@ -27,4 +27,4 @@ public class IOUringSocketDataReadInitialStateTest extends SocketDataReadInitial
protected List<TestsuitePermutation.BootstrapComboFactory<ServerBootstrap, Bootstrap>> newFactories() {
return IOUringSocketTestPermutation.INSTANCE.socket();
}
}
}

2
transport-native-io_uring/src/test/java/io/netty/channel/uring/IOUringSocketExceptionHandlingTest.java
View File

@ -27,4 +27,4 @@ public class IOUringSocketExceptionHandlingTest extends SocketExceptionHandlingT
protected List<TestsuitePermutation.BootstrapComboFactory<ServerBootstrap, Bootstrap>> newFactories() {
return IOUringSocketTestPermutation.INSTANCE.socket();
}
}
}

37
transport-native-io_uring/src/test/java/io/netty/channel/uring/IOUringSubmissionQueueTest.java
View File

@ -1,3 +1,18 @@
/*
* Copyright 2020 The Netty Project
*
* The Netty Project licenses this file to you 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 io.netty.channel.uring;
import org.junit.BeforeClass;
@ -17,17 +32,21 @@ public class IOUringSubmissionQueueTest {
@Test
public void sqeFullTest() {
RingBuffer ringBuffer = Native.createRingBuffer(8);
IOUringSubmissionQueue submissionQueue = ringBuffer.getIoUringSubmissionQueue();
final IOUringCompletionQueue completionQueue = ringBuffer.getIoUringCompletionQueue();
try {
IOUringSubmissionQueue submissionQueue = ringBuffer.getIoUringSubmissionQueue();
final IOUringCompletionQueue completionQueue = ringBuffer.getIoUringCompletionQueue();
assertNotNull(ringBuffer);
assertNotNull(submissionQueue);
assertNotNull(completionQueue);
assertNotNull(ringBuffer);
assertNotNull(submissionQueue);
assertNotNull(completionQueue);
int counter = 0;
while(!submissionQueue.addAccept(-1)) {
counter++;
int counter = 0;
while (!submissionQueue.addAccept(-1)) {
counter++;
}
assertEquals(8, counter);
} finally {
ringBuffer.close();
}
assertEquals(8, counter);
}
}

55
transport-native-io_uring/src/test/java/io/netty/channel/uring/PollRemoveTest.java
View File

@ -1,12 +1,23 @@
/*
* Copyright 2020 The Netty Project
*
* The Netty Project licenses this file to you 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 io.netty.channel.uring;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.socket.ServerSocketChannel;
import io.netty.channel.socket.SocketChannel;
@ -24,29 +35,7 @@ public class PollRemoveTest {
assumeTrue(IOUring.isAvailable());
}
@Sharable
private static final class EchoUringServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
ctx.write(msg);
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) {
ctx.flush();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
// Close the connection when an exception is raised.
cause.printStackTrace();
ctx.close();
}
}
void io_uring_test() throws Exception {
private void io_uring_test() throws Exception {
Class<? extends ServerSocketChannel> clazz = IOUringServerSocketChannel.class;
final EventLoopGroup bossGroup = new IOUringEventLoopGroup(1);
final EventLoopGroup workerGroup = new IOUringEventLoopGroup(1);
@ -58,11 +47,7 @@ public class PollRemoveTest {
.handler(new LoggingHandler(LogLevel.TRACE))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
p.addLast(new EchoUringServerHandler());
}
public void initChannel(SocketChannel ch) { }
});
Channel sc = b.bind(2020).sync().channel();
@ -74,18 +59,12 @@ public class PollRemoveTest {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
@Test
public void test() throws Exception {
io_uring_test();
System.out.println("io_uring --------------------------------");
io_uring_test();
}
}