Fix a bug which lead to only use two threads for all tasks all the time, even if the WorkerPool contained more. See #240
This commit is contained in:
parent
4c0a5886ca
commit
f88cd3120d
@ -42,14 +42,15 @@ final class NioServerSocketChannel extends AbstractServerChannel
|
||||
final ServerSocketChannel socket;
|
||||
final Lock shutdownLock = new ReentrantLock();
|
||||
final NioWorker worker;
|
||||
final WorkerPool<NioWorker> workers;
|
||||
|
||||
|
||||
private final ServerSocketChannelConfig config;
|
||||
|
||||
static NioServerSocketChannel create(ChannelFactory factory,
|
||||
ChannelPipeline pipeline, ChannelSink sink, NioWorker worker) {
|
||||
ChannelPipeline pipeline, ChannelSink sink, NioWorker worker, WorkerPool<NioWorker> workers) {
|
||||
NioServerSocketChannel instance =
|
||||
new NioServerSocketChannel(factory, pipeline, sink, worker);
|
||||
new NioServerSocketChannel(factory, pipeline, sink, worker, workers);
|
||||
fireChannelOpen(instance);
|
||||
return instance;
|
||||
}
|
||||
@ -57,10 +58,11 @@ final class NioServerSocketChannel extends AbstractServerChannel
|
||||
private NioServerSocketChannel(
|
||||
ChannelFactory factory,
|
||||
ChannelPipeline pipeline,
|
||||
ChannelSink sink, NioWorker worker) {
|
||||
ChannelSink sink, NioWorker worker, WorkerPool<NioWorker> workers) {
|
||||
|
||||
super(factory, pipeline, sink);
|
||||
this.worker = worker;
|
||||
this.workers = workers;
|
||||
try {
|
||||
socket = ServerSocketChannel.open();
|
||||
} catch (IOException e) {
|
||||
|
@ -137,13 +137,13 @@ public class NioServerSocketChannelFactory implements ServerSocketChannelFactory
|
||||
}
|
||||
|
||||
this.workerPool = workerPool;
|
||||
sink = new NioServerSocketPipelineSink(workerPool);
|
||||
sink = new NioServerSocketPipelineSink();
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public ServerSocketChannel newChannel(ChannelPipeline pipeline) {
|
||||
return NioServerSocketChannel.create(this, pipeline, sink, workerPool.nextWorker());
|
||||
return NioServerSocketChannel.create(this, pipeline, sink, workerPool.nextWorker(), workerPool);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
@ -35,12 +35,6 @@ class NioServerSocketPipelineSink extends AbstractNioChannelSink {
|
||||
static final InternalLogger logger =
|
||||
InternalLoggerFactory.getInstance(NioServerSocketPipelineSink.class);
|
||||
|
||||
private final WorkerPool<NioWorker> workerPool;
|
||||
|
||||
NioServerSocketPipelineSink(WorkerPool<NioWorker> workerPool) {
|
||||
this.workerPool = workerPool;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void eventSunk(
|
||||
ChannelPipeline pipeline, ChannelEvent e) throws Exception {
|
||||
@ -124,7 +118,7 @@ class NioServerSocketPipelineSink extends AbstractNioChannelSink {
|
||||
future.setSuccess();
|
||||
fireChannelBound(channel, channel.getLocalAddress());
|
||||
|
||||
workerPool.nextWorker().registerWithWorker(channel, future);
|
||||
channel.getWorker().registerWithWorker(channel, future);
|
||||
|
||||
} catch (Throwable t) {
|
||||
future.setFailure(t);
|
||||
|
@ -127,8 +127,7 @@ public class NioWorker extends SelectorEventLoop {
|
||||
boolean registered = channel.getJdkChannel().isRegistered();
|
||||
if (!registered) {
|
||||
synchronized (channel.interestOpsLock) {
|
||||
channel.getJdkChannel().register(
|
||||
selector, channel.getRawInterestOps(), channel);
|
||||
channel.getJdkChannel().register(selector, channel.getRawInterestOps(), channel);
|
||||
}
|
||||
|
||||
} else {
|
||||
|
@ -346,8 +346,10 @@ abstract class SelectorEventLoop extends SingleThreadEventLoop {
|
||||
// TODO: Remove the casting stuff
|
||||
ChannelPipeline pipeline =
|
||||
channel.getConfig().getPipelineFactory().getPipeline();
|
||||
registerTask(NioAcceptedSocketChannel.create(channel.getFactory(), pipeline, channel,
|
||||
channel.getPipeline().getSink(), acceptedSocket, (NioWorker) this), null);
|
||||
NioWorker worker = channel.workers.nextWorker();
|
||||
|
||||
worker.registerWithWorker(NioAcceptedSocketChannel.create(channel.getFactory(), pipeline, channel,
|
||||
channel.getPipeline().getSink(), acceptedSocket, worker), null);
|
||||
handled = true;
|
||||
}
|
||||
return handled;
|
||||
|
Loading…
Reference in New Issue
Block a user