Simplify the construction of multi-threaded selector event loop
- Hide InternalLogger from users
This commit is contained in:
parent
311f17f6ef
commit
f00fadb9fd
|
@ -22,11 +22,10 @@ import io.netty.channel.ChannelInitializer;
|
||||||
import io.netty.channel.ChannelOption;
|
import io.netty.channel.ChannelOption;
|
||||||
import io.netty.channel.ChannelPipeline;
|
import io.netty.channel.ChannelPipeline;
|
||||||
import io.netty.channel.EventLoop;
|
import io.netty.channel.EventLoop;
|
||||||
import io.netty.channel.MultithreadEventLoop;
|
|
||||||
import io.netty.channel.socket.nio.NioSocketChannel;
|
import io.netty.channel.socket.nio.NioSocketChannel;
|
||||||
import io.netty.channel.socket.nio.SelectorEventLoop;
|
import io.netty.channel.socket.nio.SelectorEventLoop;
|
||||||
|
import io.netty.handler.logging.LogLevel;
|
||||||
import io.netty.handler.logging.LoggingHandler;
|
import io.netty.handler.logging.LoggingHandler;
|
||||||
import io.netty.logging.InternalLogLevel;
|
|
||||||
|
|
||||||
import java.net.InetSocketAddress;
|
import java.net.InetSocketAddress;
|
||||||
|
|
||||||
|
@ -50,7 +49,7 @@ public class EchoClient {
|
||||||
|
|
||||||
public void run() throws Exception {
|
public void run() throws Exception {
|
||||||
// Create the required event loop.
|
// Create the required event loop.
|
||||||
EventLoop loop = new MultithreadEventLoop(SelectorEventLoop.FACTORY);
|
EventLoop loop = new SelectorEventLoop();
|
||||||
try {
|
try {
|
||||||
// Configure the client.
|
// Configure the client.
|
||||||
ChannelBuilder b = new ChannelBuilder();
|
ChannelBuilder b = new ChannelBuilder();
|
||||||
|
@ -62,7 +61,7 @@ public class EchoClient {
|
||||||
@Override
|
@Override
|
||||||
public void initChannel(Channel ch) throws Exception {
|
public void initChannel(Channel ch) throws Exception {
|
||||||
ChannelPipeline p = ch.pipeline();
|
ChannelPipeline p = ch.pipeline();
|
||||||
p.addLast("logger", new LoggingHandler(InternalLogLevel.INFO));
|
p.addLast("logger", new LoggingHandler(LogLevel.INFO));
|
||||||
p.addLast("echoer", new EchoClientHandler(firstMessageSize));
|
p.addLast("echoer", new EchoClientHandler(firstMessageSize));
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
|
@ -21,12 +21,11 @@ import io.netty.channel.ChannelInitializer;
|
||||||
import io.netty.channel.ChannelOption;
|
import io.netty.channel.ChannelOption;
|
||||||
import io.netty.channel.ChannelPipeline;
|
import io.netty.channel.ChannelPipeline;
|
||||||
import io.netty.channel.EventLoop;
|
import io.netty.channel.EventLoop;
|
||||||
import io.netty.channel.MultithreadEventLoop;
|
|
||||||
import io.netty.channel.ServerChannelBuilder;
|
import io.netty.channel.ServerChannelBuilder;
|
||||||
import io.netty.channel.socket.nio.NioServerSocketChannel;
|
import io.netty.channel.socket.nio.NioServerSocketChannel;
|
||||||
import io.netty.channel.socket.nio.SelectorEventLoop;
|
import io.netty.channel.socket.nio.SelectorEventLoop;
|
||||||
|
import io.netty.handler.logging.LogLevel;
|
||||||
import io.netty.handler.logging.LoggingHandler;
|
import io.netty.handler.logging.LoggingHandler;
|
||||||
import io.netty.logging.InternalLogLevel;
|
|
||||||
|
|
||||||
import java.net.InetSocketAddress;
|
import java.net.InetSocketAddress;
|
||||||
|
|
||||||
|
@ -43,21 +42,22 @@ public class EchoServer {
|
||||||
|
|
||||||
public void run() throws Exception {
|
public void run() throws Exception {
|
||||||
// Create the required event loops.
|
// Create the required event loops.
|
||||||
EventLoop parentLoop = new MultithreadEventLoop(SelectorEventLoop.FACTORY);
|
EventLoop parentLoop = new SelectorEventLoop();
|
||||||
EventLoop childLoop = new MultithreadEventLoop(SelectorEventLoop.FACTORY);
|
EventLoop childLoop = new SelectorEventLoop();
|
||||||
try {
|
try {
|
||||||
// Configure the server.
|
// Configure the server.
|
||||||
ServerChannelBuilder b = new ServerChannelBuilder();
|
ServerChannelBuilder b = new ServerChannelBuilder();
|
||||||
b.parentEventLoop(parentLoop)
|
b.parentEventLoop(parentLoop)
|
||||||
.childEventLoop(childLoop)
|
|
||||||
.parentChannel(new NioServerSocketChannel())
|
.parentChannel(new NioServerSocketChannel())
|
||||||
.childOption(ChannelOption.TCP_NODELAY, true)
|
.parentOption(ChannelOption.SO_BACKLOG, 24)
|
||||||
.localAddress(new InetSocketAddress(port))
|
.localAddress(new InetSocketAddress(port))
|
||||||
|
.childEventLoop(childLoop)
|
||||||
|
.childOption(ChannelOption.TCP_NODELAY, true)
|
||||||
.childInitializer(new ChannelInitializer() {
|
.childInitializer(new ChannelInitializer() {
|
||||||
@Override
|
@Override
|
||||||
public void initChannel(Channel ch) throws Exception {
|
public void initChannel(Channel ch) throws Exception {
|
||||||
ChannelPipeline p = ch.pipeline();
|
ChannelPipeline p = ch.pipeline();
|
||||||
p.addLast("logger", new LoggingHandler(InternalLogLevel.INFO));
|
p.addLast("logger", new LoggingHandler(LogLevel.INFO));
|
||||||
p.addLast("echoer", new EchoServerHandler());
|
p.addLast("echoer", new EchoServerHandler());
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
20
handler/src/main/java/io/netty/handler/logging/LogLevel.java
Normal file
20
handler/src/main/java/io/netty/handler/logging/LogLevel.java
Normal file
|
@ -0,0 +1,20 @@
|
||||||
|
package io.netty.handler.logging;
|
||||||
|
|
||||||
|
import io.netty.logging.InternalLogLevel;
|
||||||
|
|
||||||
|
public enum LogLevel {
|
||||||
|
DEBUG(InternalLogLevel.DEBUG),
|
||||||
|
INFO(InternalLogLevel.INFO),
|
||||||
|
WARN(InternalLogLevel.WARN),
|
||||||
|
ERROR(InternalLogLevel.ERROR);
|
||||||
|
|
||||||
|
private final InternalLogLevel internalLevel;
|
||||||
|
|
||||||
|
LogLevel(InternalLogLevel internalLevel) {
|
||||||
|
this.internalLevel = internalLevel;
|
||||||
|
}
|
||||||
|
|
||||||
|
InternalLogLevel toInternalLevel() {
|
||||||
|
return internalLevel;
|
||||||
|
}
|
||||||
|
}
|
|
@ -42,7 +42,7 @@ import java.util.Queue;
|
||||||
@Sharable
|
@Sharable
|
||||||
public class LoggingHandler extends ChannelHandlerAdapter<Object, Object> {
|
public class LoggingHandler extends ChannelHandlerAdapter<Object, Object> {
|
||||||
|
|
||||||
private static final InternalLogLevel DEFAULT_LEVEL = InternalLogLevel.DEBUG;
|
private static final LogLevel DEFAULT_LEVEL = LogLevel.DEBUG;
|
||||||
private static final String NEWLINE = String.format("%n");
|
private static final String NEWLINE = String.format("%n");
|
||||||
|
|
||||||
private static final String[] BYTE2HEX = new String[256];
|
private static final String[] BYTE2HEX = new String[256];
|
||||||
|
@ -104,7 +104,8 @@ public class LoggingHandler extends ChannelHandlerAdapter<Object, Object> {
|
||||||
}
|
}
|
||||||
|
|
||||||
private final InternalLogger logger;
|
private final InternalLogger logger;
|
||||||
private final InternalLogLevel level;
|
private final LogLevel level;
|
||||||
|
private final InternalLogLevel internalLevel;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Creates a new instance whose logger name is the fully qualified class
|
* Creates a new instance whose logger name is the fully qualified class
|
||||||
|
@ -120,13 +121,14 @@ public class LoggingHandler extends ChannelHandlerAdapter<Object, Object> {
|
||||||
*
|
*
|
||||||
* @param level the log level
|
* @param level the log level
|
||||||
*/
|
*/
|
||||||
public LoggingHandler(InternalLogLevel level) {
|
public LoggingHandler(LogLevel level) {
|
||||||
if (level == null) {
|
if (level == null) {
|
||||||
throw new NullPointerException("level");
|
throw new NullPointerException("level");
|
||||||
}
|
}
|
||||||
|
|
||||||
logger = InternalLoggerFactory.getInstance(getClass());
|
logger = InternalLoggerFactory.getInstance(getClass());
|
||||||
this.level = level;
|
this.level = level;
|
||||||
|
internalLevel = level.toInternalLevel();
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
@ -142,7 +144,7 @@ public class LoggingHandler extends ChannelHandlerAdapter<Object, Object> {
|
||||||
*
|
*
|
||||||
* @param level the log level
|
* @param level the log level
|
||||||
*/
|
*/
|
||||||
public LoggingHandler(Class<?> clazz, InternalLogLevel level) {
|
public LoggingHandler(Class<?> clazz, LogLevel level) {
|
||||||
if (clazz == null) {
|
if (clazz == null) {
|
||||||
throw new NullPointerException("clazz");
|
throw new NullPointerException("clazz");
|
||||||
}
|
}
|
||||||
|
@ -151,34 +153,22 @@ public class LoggingHandler extends ChannelHandlerAdapter<Object, Object> {
|
||||||
}
|
}
|
||||||
logger = InternalLoggerFactory.getInstance(clazz);
|
logger = InternalLoggerFactory.getInstance(clazz);
|
||||||
this.level = level;
|
this.level = level;
|
||||||
}
|
internalLevel = level.toInternalLevel();
|
||||||
|
|
||||||
/**
|
|
||||||
* Creates a new instance with the specified logger name and with hex dump
|
|
||||||
* enabled.
|
|
||||||
*/
|
|
||||||
public LoggingHandler(String name) {
|
|
||||||
this(name, true);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Creates a new instance with the specified logger name.
|
* Creates a new instance with the specified logger name.
|
||||||
*
|
|
||||||
* @param hexDump {@code true} if and only if the hex dump of the received
|
|
||||||
* message is logged
|
|
||||||
*/
|
*/
|
||||||
public LoggingHandler(String name, boolean hexDump) {
|
public LoggingHandler(String name) {
|
||||||
this(name, DEFAULT_LEVEL, hexDump);
|
this(name, DEFAULT_LEVEL);
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Creates a new instance with the specified logger name.
|
* Creates a new instance with the specified logger name.
|
||||||
*
|
*
|
||||||
* @param level the log level
|
* @param level the log level
|
||||||
* @param hexDump {@code true} if and only if the hex dump of the received
|
|
||||||
* message is logged
|
|
||||||
*/
|
*/
|
||||||
public LoggingHandler(String name, InternalLogLevel level, boolean hexDump) {
|
public LoggingHandler(String name, LogLevel level) {
|
||||||
if (name == null) {
|
if (name == null) {
|
||||||
throw new NullPointerException("name");
|
throw new NullPointerException("name");
|
||||||
}
|
}
|
||||||
|
@ -187,6 +177,7 @@ public class LoggingHandler extends ChannelHandlerAdapter<Object, Object> {
|
||||||
}
|
}
|
||||||
logger = InternalLoggerFactory.getInstance(name);
|
logger = InternalLoggerFactory.getInstance(name);
|
||||||
this.level = level;
|
this.level = level;
|
||||||
|
internalLevel = level.toInternalLevel();
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
@ -201,7 +192,7 @@ public class LoggingHandler extends ChannelHandlerAdapter<Object, Object> {
|
||||||
* Returns the {@link InternalLogLevel} that this handler uses to log
|
* Returns the {@link InternalLogLevel} that this handler uses to log
|
||||||
* a {@link ChannelEvent}.
|
* a {@link ChannelEvent}.
|
||||||
*/
|
*/
|
||||||
public InternalLogLevel getLevel() {
|
public LogLevel getLevel() {
|
||||||
return level;
|
return level;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -306,8 +297,8 @@ public class LoggingHandler extends ChannelHandlerAdapter<Object, Object> {
|
||||||
@Override
|
@Override
|
||||||
public void channelRegistered(ChannelInboundHandlerContext<Object> ctx)
|
public void channelRegistered(ChannelInboundHandlerContext<Object> ctx)
|
||||||
throws Exception {
|
throws Exception {
|
||||||
if (getLogger().isEnabled(level)) {
|
if (getLogger().isEnabled(internalLevel)) {
|
||||||
logger.log(level, format(ctx, "REGISTERED"));
|
logger.log(internalLevel, format(ctx, "REGISTERED"));
|
||||||
}
|
}
|
||||||
super.channelRegistered(ctx);
|
super.channelRegistered(ctx);
|
||||||
}
|
}
|
||||||
|
@ -315,8 +306,8 @@ public class LoggingHandler extends ChannelHandlerAdapter<Object, Object> {
|
||||||
@Override
|
@Override
|
||||||
public void channelUnregistered(ChannelInboundHandlerContext<Object> ctx)
|
public void channelUnregistered(ChannelInboundHandlerContext<Object> ctx)
|
||||||
throws Exception {
|
throws Exception {
|
||||||
if (getLogger().isEnabled(level)) {
|
if (getLogger().isEnabled(internalLevel)) {
|
||||||
logger.log(level, format(ctx, "UNREGISTERED"));
|
logger.log(internalLevel, format(ctx, "UNREGISTERED"));
|
||||||
}
|
}
|
||||||
super.channelUnregistered(ctx);
|
super.channelUnregistered(ctx);
|
||||||
}
|
}
|
||||||
|
@ -324,8 +315,8 @@ public class LoggingHandler extends ChannelHandlerAdapter<Object, Object> {
|
||||||
@Override
|
@Override
|
||||||
public void channelActive(ChannelInboundHandlerContext<Object> ctx)
|
public void channelActive(ChannelInboundHandlerContext<Object> ctx)
|
||||||
throws Exception {
|
throws Exception {
|
||||||
if (getLogger().isEnabled(level)) {
|
if (getLogger().isEnabled(internalLevel)) {
|
||||||
logger.log(level, format(ctx, "ACTIVE"));
|
logger.log(internalLevel, format(ctx, "ACTIVE"));
|
||||||
}
|
}
|
||||||
super.channelActive(ctx);
|
super.channelActive(ctx);
|
||||||
}
|
}
|
||||||
|
@ -333,8 +324,8 @@ public class LoggingHandler extends ChannelHandlerAdapter<Object, Object> {
|
||||||
@Override
|
@Override
|
||||||
public void channelInactive(ChannelInboundHandlerContext<Object> ctx)
|
public void channelInactive(ChannelInboundHandlerContext<Object> ctx)
|
||||||
throws Exception {
|
throws Exception {
|
||||||
if (getLogger().isEnabled(level)) {
|
if (getLogger().isEnabled(internalLevel)) {
|
||||||
logger.log(level, format(ctx, "INACTIVE"));
|
logger.log(internalLevel, format(ctx, "INACTIVE"));
|
||||||
}
|
}
|
||||||
super.channelInactive(ctx);
|
super.channelInactive(ctx);
|
||||||
}
|
}
|
||||||
|
@ -342,8 +333,8 @@ public class LoggingHandler extends ChannelHandlerAdapter<Object, Object> {
|
||||||
@Override
|
@Override
|
||||||
public void exceptionCaught(ChannelInboundHandlerContext<Object> ctx,
|
public void exceptionCaught(ChannelInboundHandlerContext<Object> ctx,
|
||||||
Throwable cause) throws Exception {
|
Throwable cause) throws Exception {
|
||||||
if (getLogger().isEnabled(level)) {
|
if (getLogger().isEnabled(internalLevel)) {
|
||||||
logger.log(level, format(ctx, "EXCEPTION: " + cause), cause);
|
logger.log(internalLevel, format(ctx, "EXCEPTION: " + cause), cause);
|
||||||
}
|
}
|
||||||
super.exceptionCaught(ctx, cause);
|
super.exceptionCaught(ctx, cause);
|
||||||
}
|
}
|
||||||
|
@ -351,8 +342,8 @@ public class LoggingHandler extends ChannelHandlerAdapter<Object, Object> {
|
||||||
@Override
|
@Override
|
||||||
public void userEventTriggered(ChannelInboundHandlerContext<Object> ctx,
|
public void userEventTriggered(ChannelInboundHandlerContext<Object> ctx,
|
||||||
Object evt) throws Exception {
|
Object evt) throws Exception {
|
||||||
if (getLogger().isEnabled(level)) {
|
if (getLogger().isEnabled(internalLevel)) {
|
||||||
logger.log(level, format(ctx, "USER_EVENT: " + evt));
|
logger.log(internalLevel, format(ctx, "USER_EVENT: " + evt));
|
||||||
}
|
}
|
||||||
super.userEventTriggered(ctx, evt);
|
super.userEventTriggered(ctx, evt);
|
||||||
}
|
}
|
||||||
|
@ -360,8 +351,8 @@ public class LoggingHandler extends ChannelHandlerAdapter<Object, Object> {
|
||||||
@Override
|
@Override
|
||||||
public void inboundBufferUpdated(ChannelInboundHandlerContext<Object> ctx)
|
public void inboundBufferUpdated(ChannelInboundHandlerContext<Object> ctx)
|
||||||
throws Exception {
|
throws Exception {
|
||||||
if (getLogger().isEnabled(level)) {
|
if (getLogger().isEnabled(internalLevel)) {
|
||||||
logger.log(level, format(ctx, formatBuffer("INBUF", ctx.in())));
|
logger.log(internalLevel, format(ctx, formatBuffer("INBUF", ctx.in())));
|
||||||
}
|
}
|
||||||
ctx.fireInboundBufferUpdated();
|
ctx.fireInboundBufferUpdated();
|
||||||
}
|
}
|
||||||
|
@ -369,8 +360,8 @@ public class LoggingHandler extends ChannelHandlerAdapter<Object, Object> {
|
||||||
@Override
|
@Override
|
||||||
public void bind(ChannelOutboundHandlerContext<Object> ctx,
|
public void bind(ChannelOutboundHandlerContext<Object> ctx,
|
||||||
SocketAddress localAddress, ChannelFuture future) throws Exception {
|
SocketAddress localAddress, ChannelFuture future) throws Exception {
|
||||||
if (getLogger().isEnabled(level)) {
|
if (getLogger().isEnabled(internalLevel)) {
|
||||||
logger.log(level, format(ctx, "BIND(" + localAddress + ')'));
|
logger.log(internalLevel, format(ctx, "BIND(" + localAddress + ')'));
|
||||||
}
|
}
|
||||||
super.bind(ctx, localAddress, future);
|
super.bind(ctx, localAddress, future);
|
||||||
}
|
}
|
||||||
|
@ -379,8 +370,8 @@ public class LoggingHandler extends ChannelHandlerAdapter<Object, Object> {
|
||||||
public void connect(ChannelOutboundHandlerContext<Object> ctx,
|
public void connect(ChannelOutboundHandlerContext<Object> ctx,
|
||||||
SocketAddress remoteAddress, SocketAddress localAddress,
|
SocketAddress remoteAddress, SocketAddress localAddress,
|
||||||
ChannelFuture future) throws Exception {
|
ChannelFuture future) throws Exception {
|
||||||
if (getLogger().isEnabled(level)) {
|
if (getLogger().isEnabled(internalLevel)) {
|
||||||
logger.log(level, format(ctx, "CONNECT(" + remoteAddress + ", " + localAddress + ')'));
|
logger.log(internalLevel, format(ctx, "CONNECT(" + remoteAddress + ", " + localAddress + ')'));
|
||||||
}
|
}
|
||||||
super.connect(ctx, remoteAddress, localAddress, future);
|
super.connect(ctx, remoteAddress, localAddress, future);
|
||||||
}
|
}
|
||||||
|
@ -388,8 +379,8 @@ public class LoggingHandler extends ChannelHandlerAdapter<Object, Object> {
|
||||||
@Override
|
@Override
|
||||||
public void disconnect(ChannelOutboundHandlerContext<Object> ctx,
|
public void disconnect(ChannelOutboundHandlerContext<Object> ctx,
|
||||||
ChannelFuture future) throws Exception {
|
ChannelFuture future) throws Exception {
|
||||||
if (getLogger().isEnabled(level)) {
|
if (getLogger().isEnabled(internalLevel)) {
|
||||||
logger.log(level, format(ctx, "DISCONNECT()"));
|
logger.log(internalLevel, format(ctx, "DISCONNECT()"));
|
||||||
}
|
}
|
||||||
super.disconnect(ctx, future);
|
super.disconnect(ctx, future);
|
||||||
}
|
}
|
||||||
|
@ -397,8 +388,8 @@ public class LoggingHandler extends ChannelHandlerAdapter<Object, Object> {
|
||||||
@Override
|
@Override
|
||||||
public void close(ChannelOutboundHandlerContext<Object> ctx,
|
public void close(ChannelOutboundHandlerContext<Object> ctx,
|
||||||
ChannelFuture future) throws Exception {
|
ChannelFuture future) throws Exception {
|
||||||
if (getLogger().isEnabled(level)) {
|
if (getLogger().isEnabled(internalLevel)) {
|
||||||
logger.log(level, format(ctx, "CLOSE()"));
|
logger.log(internalLevel, format(ctx, "CLOSE()"));
|
||||||
}
|
}
|
||||||
super.close(ctx, future);
|
super.close(ctx, future);
|
||||||
}
|
}
|
||||||
|
@ -406,8 +397,8 @@ public class LoggingHandler extends ChannelHandlerAdapter<Object, Object> {
|
||||||
@Override
|
@Override
|
||||||
public void deregister(ChannelOutboundHandlerContext<Object> ctx,
|
public void deregister(ChannelOutboundHandlerContext<Object> ctx,
|
||||||
ChannelFuture future) throws Exception {
|
ChannelFuture future) throws Exception {
|
||||||
if (getLogger().isEnabled(level)) {
|
if (getLogger().isEnabled(internalLevel)) {
|
||||||
logger.log(level, format(ctx, "DEREGISTER()"));
|
logger.log(internalLevel, format(ctx, "DEREGISTER()"));
|
||||||
}
|
}
|
||||||
super.deregister(ctx, future);
|
super.deregister(ctx, future);
|
||||||
}
|
}
|
||||||
|
@ -415,8 +406,8 @@ public class LoggingHandler extends ChannelHandlerAdapter<Object, Object> {
|
||||||
@Override
|
@Override
|
||||||
public void flush(ChannelOutboundHandlerContext<Object> ctx,
|
public void flush(ChannelOutboundHandlerContext<Object> ctx,
|
||||||
ChannelFuture future) throws Exception {
|
ChannelFuture future) throws Exception {
|
||||||
if (getLogger().isEnabled(level)) {
|
if (getLogger().isEnabled(internalLevel)) {
|
||||||
logger.log(level, format(ctx, formatBuffer("OUTBUF", ctx.prevOut())));
|
logger.log(internalLevel, format(ctx, formatBuffer("OUTBUF", ctx.prevOut())));
|
||||||
}
|
}
|
||||||
ctx.flush(future);
|
ctx.flush(future);
|
||||||
}
|
}
|
||||||
|
|
|
@ -43,7 +43,6 @@ public abstract class ChannelInitializer extends ChannelInboundHandlerAdapter<Ob
|
||||||
ctx.pipeline().remove(this);
|
ctx.pipeline().remove(this);
|
||||||
// Note that we do not call ctx.fireChannelRegistered() because a user might have
|
// Note that we do not call ctx.fireChannelRegistered() because a user might have
|
||||||
// inserted a handler before the initializer using pipeline.addFirst().
|
// inserted a handler before the initializer using pipeline.addFirst().
|
||||||
System.out.println(ctx.pipeline().toMap());
|
|
||||||
ctx.pipeline().fireChannelRegistered();
|
ctx.pipeline().fireChannelRegistered();
|
||||||
} catch (Throwable t) {
|
} catch (Throwable t) {
|
||||||
logger.warn("Failed to initialize a channel. Closing: " + ctx.channel());
|
logger.warn("Failed to initialize a channel. Closing: " + ctx.channel());
|
||||||
|
|
|
@ -15,15 +15,18 @@ import java.util.concurrent.atomic.AtomicInteger;
|
||||||
|
|
||||||
public class MultithreadEventLoop implements EventLoop {
|
public class MultithreadEventLoop implements EventLoop {
|
||||||
|
|
||||||
|
protected static final int DEFAULT_POOL_SIZE = Runtime.getRuntime().availableProcessors() * 2;
|
||||||
|
protected static final ThreadFactory DEFAULT_THREAD_FACTORY = Executors.defaultThreadFactory();
|
||||||
|
|
||||||
private final EventLoop[] children;
|
private final EventLoop[] children;
|
||||||
private final AtomicInteger childIndex = new AtomicInteger();
|
private final AtomicInteger childIndex = new AtomicInteger();
|
||||||
|
|
||||||
public MultithreadEventLoop(EventLoopFactory<? extends SingleThreadEventLoop> loopFactory) {
|
public MultithreadEventLoop(EventLoopFactory<? extends SingleThreadEventLoop> loopFactory) {
|
||||||
this(loopFactory, Runtime.getRuntime().availableProcessors() * 2);
|
this(loopFactory, DEFAULT_POOL_SIZE);
|
||||||
}
|
}
|
||||||
|
|
||||||
public MultithreadEventLoop(EventLoopFactory<? extends SingleThreadEventLoop> loopFactory, int nThreads) {
|
public MultithreadEventLoop(EventLoopFactory<? extends SingleThreadEventLoop> loopFactory, int nThreads) {
|
||||||
this(loopFactory, nThreads, Executors.defaultThreadFactory());
|
this(loopFactory, nThreads, DEFAULT_THREAD_FACTORY);
|
||||||
}
|
}
|
||||||
|
|
||||||
public MultithreadEventLoop(EventLoopFactory<? extends SingleThreadEventLoop> loopFactory, int nThreads, ThreadFactory threadFactory) {
|
public MultithreadEventLoop(EventLoopFactory<? extends SingleThreadEventLoop> loopFactory, int nThreads, ThreadFactory threadFactory) {
|
||||||
|
|
|
@ -37,11 +37,6 @@ public abstract class AbstractNioChannel extends AbstractChannel {
|
||||||
this.ch = ch;
|
this.ch = ch;
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
|
||||||
public SelectorEventLoop eventLoop() {
|
|
||||||
return (SelectorEventLoop) super.eventLoop();
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
protected SelectableChannel javaChannel() {
|
protected SelectableChannel javaChannel() {
|
||||||
return ch;
|
return ch;
|
||||||
|
@ -84,12 +79,12 @@ public abstract class AbstractNioChannel extends AbstractChannel {
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
protected boolean isCompatible(EventLoop loop) {
|
protected boolean isCompatible(EventLoop loop) {
|
||||||
return loop instanceof SelectorEventLoop;
|
return loop instanceof SingleThreadSelectorEventLoop;
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
protected void doRegister() throws Exception {
|
protected void doRegister() throws Exception {
|
||||||
SelectorEventLoop loop = eventLoop();
|
SingleThreadSelectorEventLoop loop = (SingleThreadSelectorEventLoop) eventLoop();
|
||||||
selectionKey = javaChannel().register(loop.selector, isActive()? SelectionKey.OP_READ : 0, this);
|
selectionKey = javaChannel().register(loop.selector, isActive()? SelectionKey.OP_READ : 0, this);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -43,7 +43,7 @@ import java.util.concurrent.Executor;
|
||||||
* A class responsible for registering channels with {@link Selector}.
|
* A class responsible for registering channels with {@link Selector}.
|
||||||
* It also implements the {@link Selector} loop.
|
* It also implements the {@link Selector} loop.
|
||||||
*/
|
*/
|
||||||
public class NioDatagramWorker extends SelectorEventLoop {
|
public class NioDatagramWorker extends SingleThreadSelectorEventLoop {
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Sole constructor.
|
* Sole constructor.
|
||||||
|
|
|
@ -74,11 +74,6 @@ public class NioServerSocketChannel extends AbstractServerChannel
|
||||||
return config;
|
return config;
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
|
||||||
public SelectorEventLoop eventLoop() {
|
|
||||||
return (SelectorEventLoop) super.eventLoop();
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public boolean isActive() {
|
public boolean isActive() {
|
||||||
return javaChannel().socket().isBound();
|
return javaChannel().socket().isBound();
|
||||||
|
@ -116,12 +111,12 @@ public class NioServerSocketChannel extends AbstractServerChannel
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
protected boolean isCompatible(EventLoop loop) {
|
protected boolean isCompatible(EventLoop loop) {
|
||||||
return loop instanceof SelectorEventLoop;
|
return loop instanceof SingleThreadSelectorEventLoop;
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
protected void doRegister() throws Exception {
|
protected void doRegister() throws Exception {
|
||||||
SelectorEventLoop loop = eventLoop();
|
SingleThreadSelectorEventLoop loop = (SingleThreadSelectorEventLoop) eventLoop();
|
||||||
selectionKey = javaChannel().register(
|
selectionKey = javaChannel().register(
|
||||||
loop.selector, isActive()? SelectionKey.OP_ACCEPT : 0, this);
|
loop.selector, isActive()? SelectionKey.OP_ACCEPT : 0, this);
|
||||||
}
|
}
|
||||||
|
|
|
@ -152,7 +152,7 @@ public class NioSocketChannel extends AbstractNioChannel implements io.netty.cha
|
||||||
@Override
|
@Override
|
||||||
protected void doDeregister() throws Exception {
|
protected void doDeregister() throws Exception {
|
||||||
selectionKey().cancel();
|
selectionKey().cancel();
|
||||||
eventLoop().cancelledKeys ++;
|
((SingleThreadSelectorEventLoop) eventLoop()).cancelledKeys ++;
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
|
|
@ -1,253 +1,32 @@
|
||||||
/*
|
|
||||||
* Copyright 2011 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.socket.nio;
|
package io.netty.channel.socket.nio;
|
||||||
|
|
||||||
import io.netty.channel.Channel;
|
|
||||||
import io.netty.channel.Channel.Unsafe;
|
|
||||||
import io.netty.channel.ChannelException;
|
|
||||||
import io.netty.channel.EventLoopFactory;
|
import io.netty.channel.EventLoopFactory;
|
||||||
import io.netty.channel.SingleThreadEventLoop;
|
import io.netty.channel.MultithreadEventLoop;
|
||||||
import io.netty.logging.InternalLogger;
|
|
||||||
import io.netty.logging.InternalLoggerFactory;
|
|
||||||
|
|
||||||
import java.io.IOException;
|
|
||||||
import java.nio.channels.CancelledKeyException;
|
|
||||||
import java.nio.channels.SelectionKey;
|
|
||||||
import java.nio.channels.Selector;
|
|
||||||
import java.nio.channels.spi.SelectorProvider;
|
import java.nio.channels.spi.SelectorProvider;
|
||||||
import java.util.ArrayList;
|
|
||||||
import java.util.Collection;
|
|
||||||
import java.util.Iterator;
|
|
||||||
import java.util.Set;
|
|
||||||
import java.util.concurrent.Executors;
|
|
||||||
import java.util.concurrent.ThreadFactory;
|
import java.util.concurrent.ThreadFactory;
|
||||||
import java.util.concurrent.atomic.AtomicBoolean;
|
|
||||||
|
|
||||||
public class SelectorEventLoop extends SingleThreadEventLoop {
|
public class SelectorEventLoop extends MultithreadEventLoop {
|
||||||
|
|
||||||
public static final EventLoopFactory<SelectorEventLoop> FACTORY = new EventLoopFactory<SelectorEventLoop>() {
|
|
||||||
@Override
|
|
||||||
public SelectorEventLoop newEventLoop(ThreadFactory threadFactory)
|
|
||||||
throws Exception {
|
|
||||||
return new SelectorEventLoop(threadFactory);
|
|
||||||
}
|
|
||||||
};
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Internal Netty logger.
|
|
||||||
*/
|
|
||||||
protected static final InternalLogger logger = InternalLoggerFactory
|
|
||||||
.getInstance(SelectorEventLoop.class);
|
|
||||||
|
|
||||||
static final int CLEANUP_INTERVAL = 256; // XXX Hard-coded value, but won't need customization.
|
|
||||||
|
|
||||||
/**
|
|
||||||
* The NIO {@link Selector}.
|
|
||||||
*/
|
|
||||||
protected final Selector selector;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Boolean that controls determines if a blocked Selector.select should
|
|
||||||
* break out of its selection process. In our case we use a timeone for
|
|
||||||
* the select method and the select method will block for that time unless
|
|
||||||
* waken up.
|
|
||||||
*/
|
|
||||||
protected final AtomicBoolean wakenUp = new AtomicBoolean();
|
|
||||||
|
|
||||||
int cancelledKeys;
|
|
||||||
|
|
||||||
public SelectorEventLoop() {
|
public SelectorEventLoop() {
|
||||||
this(Executors.defaultThreadFactory());
|
this(DEFAULT_POOL_SIZE);
|
||||||
}
|
}
|
||||||
|
|
||||||
public SelectorEventLoop(ThreadFactory threadFactory) {
|
public SelectorEventLoop(int nThreads) {
|
||||||
this(threadFactory, SelectorProvider.provider());
|
this(nThreads, DEFAULT_THREAD_FACTORY);
|
||||||
}
|
}
|
||||||
|
|
||||||
public SelectorEventLoop(SelectorProvider selectorProvider) {
|
public SelectorEventLoop(int nThreads, ThreadFactory threadFactory) {
|
||||||
this(Executors.defaultThreadFactory(), selectorProvider);
|
this(nThreads, threadFactory, SelectorProvider.provider());
|
||||||
}
|
|
||||||
|
|
||||||
public SelectorEventLoop(ThreadFactory threadFactory, SelectorProvider selectorProvider) {
|
|
||||||
super(threadFactory);
|
|
||||||
if (selectorProvider == null) {
|
|
||||||
throw new NullPointerException("selectorProvider");
|
|
||||||
}
|
|
||||||
selector = openSelector(selectorProvider);
|
|
||||||
}
|
|
||||||
|
|
||||||
private static Selector openSelector(SelectorProvider provider) {
|
|
||||||
try {
|
|
||||||
return provider.openSelector();
|
|
||||||
} catch (IOException e) {
|
|
||||||
throw new ChannelException("failed to open a new selector", e);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public SelectorEventLoop(int nThreads, ThreadFactory threadFactory, final SelectorProvider selectorProvider) {
|
||||||
|
super(new EventLoopFactory<SingleThreadSelectorEventLoop>() {
|
||||||
@Override
|
@Override
|
||||||
protected void run() {
|
public SingleThreadSelectorEventLoop newEventLoop(ThreadFactory threadFactory) throws Exception {
|
||||||
Selector selector = this.selector;
|
return new SingleThreadSelectorEventLoop(threadFactory, selectorProvider);
|
||||||
for (;;) {
|
|
||||||
|
|
||||||
wakenUp.set(false);
|
|
||||||
|
|
||||||
try {
|
|
||||||
SelectorUtil.select(selector);
|
|
||||||
|
|
||||||
// 'wakenUp.compareAndSet(false, true)' is always evaluated
|
|
||||||
// before calling 'selector.wakeup()' to reduce the wake-up
|
|
||||||
// overhead. (Selector.wakeup() is an expensive operation.)
|
|
||||||
//
|
|
||||||
// However, there is a race condition in this approach.
|
|
||||||
// The race condition is triggered when 'wakenUp' is set to
|
|
||||||
// true too early.
|
|
||||||
//
|
|
||||||
// 'wakenUp' is set to true too early if:
|
|
||||||
// 1) Selector is waken up between 'wakenUp.set(false)' and
|
|
||||||
// 'selector.select(...)'. (BAD)
|
|
||||||
// 2) Selector is waken up between 'selector.select(...)' and
|
|
||||||
// 'if (wakenUp.get()) { ... }'. (OK)
|
|
||||||
//
|
|
||||||
// In the first case, 'wakenUp' is set to true and the
|
|
||||||
// following 'selector.select(...)' will wake up immediately.
|
|
||||||
// Until 'wakenUp' is set to false again in the next round,
|
|
||||||
// 'wakenUp.compareAndSet(false, true)' will fail, and therefore
|
|
||||||
// any attempt to wake up the Selector will fail, too, causing
|
|
||||||
// the following 'selector.select(...)' call to block
|
|
||||||
// unnecessarily.
|
|
||||||
//
|
|
||||||
// To fix this problem, we wake up the selector again if wakenUp
|
|
||||||
// is true immediately after selector.select(...).
|
|
||||||
// It is inefficient in that it wakes up the selector for both
|
|
||||||
// the first case (BAD - wake-up required) and the second case
|
|
||||||
// (OK - no wake-up required).
|
|
||||||
|
|
||||||
if (wakenUp.get()) {
|
|
||||||
selector.wakeup();
|
|
||||||
}
|
}
|
||||||
|
|
||||||
cancelledKeys = 0;
|
}, nThreads, threadFactory);
|
||||||
processTaskQueue();
|
|
||||||
processSelectedKeys();
|
|
||||||
|
|
||||||
if (isShutdown()) {
|
|
||||||
closeAll();
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
} catch (Throwable t) {
|
|
||||||
logger.warn(
|
|
||||||
"Unexpected exception in the selector loop.", t);
|
|
||||||
|
|
||||||
// Prevent possible consecutive immediate failures that lead to
|
|
||||||
// excessive CPU consumption.
|
|
||||||
try {
|
|
||||||
Thread.sleep(1000);
|
|
||||||
} catch (InterruptedException e) {
|
|
||||||
// Ignore.
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
protected void cleanup() {
|
|
||||||
try {
|
|
||||||
selector.close();
|
|
||||||
} catch (IOException e) {
|
|
||||||
logger.warn(
|
|
||||||
"Failed to close a selector.", e);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private void processTaskQueue() {
|
|
||||||
for (;;) {
|
|
||||||
final Runnable task = pollTask();
|
|
||||||
if (task == null) {
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
|
|
||||||
task.run();
|
|
||||||
cleanUpCancelledKeys();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private void processSelectedKeys() {
|
|
||||||
for (Iterator<SelectionKey> i = selector.selectedKeys().iterator(); i.hasNext();) {
|
|
||||||
final SelectionKey k = i.next();
|
|
||||||
final Channel ch = (Channel) k.attachment();
|
|
||||||
final Unsafe unsafe = ch.unsafe();
|
|
||||||
boolean removeKey = true;
|
|
||||||
try {
|
|
||||||
int readyOps = k.readyOps();
|
|
||||||
if ((readyOps & SelectionKey.OP_READ) != 0 || readyOps == 0) {
|
|
||||||
unsafe.read();
|
|
||||||
if (!ch.isOpen()) {
|
|
||||||
// Connection already closed - no need to handle write.
|
|
||||||
continue;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if ((readyOps & SelectionKey.OP_WRITE) != 0) {
|
|
||||||
unsafe.flushForcibly();
|
|
||||||
}
|
|
||||||
if ((readyOps & SelectionKey.OP_ACCEPT) != 0) {
|
|
||||||
unsafe.read();
|
|
||||||
}
|
|
||||||
if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
|
|
||||||
unsafe.finishConnect();
|
|
||||||
}
|
|
||||||
} catch (CancelledKeyException ignored) {
|
|
||||||
unsafe.close(unsafe.voidFuture());
|
|
||||||
} finally {
|
|
||||||
if (removeKey) {
|
|
||||||
i.remove();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if (cleanUpCancelledKeys()) {
|
|
||||||
break; // break the loop to avoid ConcurrentModificationException
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private boolean cleanUpCancelledKeys() {
|
|
||||||
if (cancelledKeys >= CLEANUP_INTERVAL) {
|
|
||||||
cancelledKeys = 0;
|
|
||||||
SelectorUtil.cleanupKeys(selector);
|
|
||||||
return true;
|
|
||||||
}
|
|
||||||
return false;
|
|
||||||
}
|
|
||||||
|
|
||||||
private void closeAll() {
|
|
||||||
SelectorUtil.cleanupKeys(selector);
|
|
||||||
Set<SelectionKey> keys = selector.keys();
|
|
||||||
Collection<Channel> channels = new ArrayList<Channel>(keys.size());
|
|
||||||
for (SelectionKey k: keys) {
|
|
||||||
channels.add((Channel) k.attachment());
|
|
||||||
}
|
|
||||||
|
|
||||||
for (Channel ch: channels) {
|
|
||||||
ch.unsafe().close(ch.unsafe().voidFuture());
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
protected void wakeup(boolean inEventLoop) {
|
|
||||||
if (wakenUp.compareAndSet(false, true)) {
|
|
||||||
selector.wakeup();
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -0,0 +1,231 @@
|
||||||
|
/*
|
||||||
|
* Copyright 2011 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.socket.nio;
|
||||||
|
|
||||||
|
import io.netty.channel.Channel;
|
||||||
|
import io.netty.channel.Channel.Unsafe;
|
||||||
|
import io.netty.channel.ChannelException;
|
||||||
|
import io.netty.channel.SingleThreadEventLoop;
|
||||||
|
import io.netty.logging.InternalLogger;
|
||||||
|
import io.netty.logging.InternalLoggerFactory;
|
||||||
|
|
||||||
|
import java.io.IOException;
|
||||||
|
import java.nio.channels.CancelledKeyException;
|
||||||
|
import java.nio.channels.SelectionKey;
|
||||||
|
import java.nio.channels.Selector;
|
||||||
|
import java.nio.channels.spi.SelectorProvider;
|
||||||
|
import java.util.ArrayList;
|
||||||
|
import java.util.Collection;
|
||||||
|
import java.util.Iterator;
|
||||||
|
import java.util.Set;
|
||||||
|
import java.util.concurrent.ThreadFactory;
|
||||||
|
import java.util.concurrent.atomic.AtomicBoolean;
|
||||||
|
|
||||||
|
final class SingleThreadSelectorEventLoop extends SingleThreadEventLoop {
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Internal Netty logger.
|
||||||
|
*/
|
||||||
|
protected static final InternalLogger logger = InternalLoggerFactory
|
||||||
|
.getInstance(SingleThreadSelectorEventLoop.class);
|
||||||
|
|
||||||
|
static final int CLEANUP_INTERVAL = 256; // XXX Hard-coded value, but won't need customization.
|
||||||
|
|
||||||
|
/**
|
||||||
|
* The NIO {@link Selector}.
|
||||||
|
*/
|
||||||
|
protected final Selector selector;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Boolean that controls determines if a blocked Selector.select should
|
||||||
|
* break out of its selection process. In our case we use a timeone for
|
||||||
|
* the select method and the select method will block for that time unless
|
||||||
|
* waken up.
|
||||||
|
*/
|
||||||
|
protected final AtomicBoolean wakenUp = new AtomicBoolean();
|
||||||
|
|
||||||
|
int cancelledKeys;
|
||||||
|
|
||||||
|
SingleThreadSelectorEventLoop(ThreadFactory threadFactory, SelectorProvider selectorProvider) {
|
||||||
|
super(threadFactory);
|
||||||
|
if (selectorProvider == null) {
|
||||||
|
throw new NullPointerException("selectorProvider");
|
||||||
|
}
|
||||||
|
selector = openSelector(selectorProvider);
|
||||||
|
}
|
||||||
|
|
||||||
|
private static Selector openSelector(SelectorProvider provider) {
|
||||||
|
try {
|
||||||
|
return provider.openSelector();
|
||||||
|
} catch (IOException e) {
|
||||||
|
throw new ChannelException("failed to open a new selector", e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
protected void run() {
|
||||||
|
Selector selector = this.selector;
|
||||||
|
for (;;) {
|
||||||
|
|
||||||
|
wakenUp.set(false);
|
||||||
|
|
||||||
|
try {
|
||||||
|
SelectorUtil.select(selector);
|
||||||
|
|
||||||
|
// 'wakenUp.compareAndSet(false, true)' is always evaluated
|
||||||
|
// before calling 'selector.wakeup()' to reduce the wake-up
|
||||||
|
// overhead. (Selector.wakeup() is an expensive operation.)
|
||||||
|
//
|
||||||
|
// However, there is a race condition in this approach.
|
||||||
|
// The race condition is triggered when 'wakenUp' is set to
|
||||||
|
// true too early.
|
||||||
|
//
|
||||||
|
// 'wakenUp' is set to true too early if:
|
||||||
|
// 1) Selector is waken up between 'wakenUp.set(false)' and
|
||||||
|
// 'selector.select(...)'. (BAD)
|
||||||
|
// 2) Selector is waken up between 'selector.select(...)' and
|
||||||
|
// 'if (wakenUp.get()) { ... }'. (OK)
|
||||||
|
//
|
||||||
|
// In the first case, 'wakenUp' is set to true and the
|
||||||
|
// following 'selector.select(...)' will wake up immediately.
|
||||||
|
// Until 'wakenUp' is set to false again in the next round,
|
||||||
|
// 'wakenUp.compareAndSet(false, true)' will fail, and therefore
|
||||||
|
// any attempt to wake up the Selector will fail, too, causing
|
||||||
|
// the following 'selector.select(...)' call to block
|
||||||
|
// unnecessarily.
|
||||||
|
//
|
||||||
|
// To fix this problem, we wake up the selector again if wakenUp
|
||||||
|
// is true immediately after selector.select(...).
|
||||||
|
// It is inefficient in that it wakes up the selector for both
|
||||||
|
// the first case (BAD - wake-up required) and the second case
|
||||||
|
// (OK - no wake-up required).
|
||||||
|
|
||||||
|
if (wakenUp.get()) {
|
||||||
|
selector.wakeup();
|
||||||
|
}
|
||||||
|
|
||||||
|
cancelledKeys = 0;
|
||||||
|
processTaskQueue();
|
||||||
|
processSelectedKeys();
|
||||||
|
|
||||||
|
if (isShutdown()) {
|
||||||
|
closeAll();
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
} catch (Throwable t) {
|
||||||
|
logger.warn(
|
||||||
|
"Unexpected exception in the selector loop.", t);
|
||||||
|
|
||||||
|
// Prevent possible consecutive immediate failures that lead to
|
||||||
|
// excessive CPU consumption.
|
||||||
|
try {
|
||||||
|
Thread.sleep(1000);
|
||||||
|
} catch (InterruptedException e) {
|
||||||
|
// Ignore.
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
protected void cleanup() {
|
||||||
|
try {
|
||||||
|
selector.close();
|
||||||
|
} catch (IOException e) {
|
||||||
|
logger.warn(
|
||||||
|
"Failed to close a selector.", e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private void processTaskQueue() {
|
||||||
|
for (;;) {
|
||||||
|
final Runnable task = pollTask();
|
||||||
|
if (task == null) {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
|
task.run();
|
||||||
|
cleanUpCancelledKeys();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private void processSelectedKeys() {
|
||||||
|
for (Iterator<SelectionKey> i = selector.selectedKeys().iterator(); i.hasNext();) {
|
||||||
|
final SelectionKey k = i.next();
|
||||||
|
final Channel ch = (Channel) k.attachment();
|
||||||
|
final Unsafe unsafe = ch.unsafe();
|
||||||
|
boolean removeKey = true;
|
||||||
|
try {
|
||||||
|
int readyOps = k.readyOps();
|
||||||
|
if ((readyOps & SelectionKey.OP_READ) != 0 || readyOps == 0) {
|
||||||
|
unsafe.read();
|
||||||
|
if (!ch.isOpen()) {
|
||||||
|
// Connection already closed - no need to handle write.
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if ((readyOps & SelectionKey.OP_WRITE) != 0) {
|
||||||
|
unsafe.flushForcibly();
|
||||||
|
}
|
||||||
|
if ((readyOps & SelectionKey.OP_ACCEPT) != 0) {
|
||||||
|
unsafe.read();
|
||||||
|
}
|
||||||
|
if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
|
||||||
|
unsafe.finishConnect();
|
||||||
|
}
|
||||||
|
} catch (CancelledKeyException ignored) {
|
||||||
|
unsafe.close(unsafe.voidFuture());
|
||||||
|
} finally {
|
||||||
|
if (removeKey) {
|
||||||
|
i.remove();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if (cleanUpCancelledKeys()) {
|
||||||
|
break; // break the loop to avoid ConcurrentModificationException
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private boolean cleanUpCancelledKeys() {
|
||||||
|
if (cancelledKeys >= CLEANUP_INTERVAL) {
|
||||||
|
cancelledKeys = 0;
|
||||||
|
SelectorUtil.cleanupKeys(selector);
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
private void closeAll() {
|
||||||
|
SelectorUtil.cleanupKeys(selector);
|
||||||
|
Set<SelectionKey> keys = selector.keys();
|
||||||
|
Collection<Channel> channels = new ArrayList<Channel>(keys.size());
|
||||||
|
for (SelectionKey k: keys) {
|
||||||
|
channels.add((Channel) k.attachment());
|
||||||
|
}
|
||||||
|
|
||||||
|
for (Channel ch: channels) {
|
||||||
|
ch.unsafe().close(ch.unsafe().voidFuture());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
protected void wakeup(boolean inEventLoop) {
|
||||||
|
if (wakenUp.compareAndSet(false, true)) {
|
||||||
|
selector.wakeup();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
Loading…
Reference in New Issue
Block a user