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

Correctly count acquired channels when timeout occurs in FixedChannelPool

Browse Source 
Motivation:

We don't decrease acquired channel count in FixedChannelPool when timeout occurs by AcquireTimeoutAction.NEW and eventually fails.

Modifications:

Set AcquireTask.acquired=true to call decrementAndRunTaskQueue when timeout action fails.

Result:

Acquired channel count decreases correctly.
This commit is contained in:
fratboy 2015-07-29 02:49:28 +09:00 committed by Norman Maurer
parent 2edcb80c49
commit b6dc1a4cec
1 changed files with 12 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
transport/src/main/java/io/netty/channel/pool/FixedChannelPool.java
View File

@ -117,7 +117,7 @@ public final class FixedChannelPool extends SimpleChannelPool {
* be failed.
*/
public FixedChannelPool(Bootstrap bootstrap,
ChannelPoolHandler handler,
ChannelPoolHandler handler,
ChannelHealthChecker healthCheck, AcquireTimeoutAction action,
final long acquireTimeoutMillis,
int maxConnections, int maxPendingAcquires) {
@ -153,7 +153,7 @@ public final class FixedChannelPool extends SimpleChannelPool {
public void onTimeout(AcquireTask task) {
// Increment the acquire count and delegate to super to actually acquire a Channel which will
// create a new connetion.
++acquiredChannelCount;
task.acquired();
FixedChannelPool.super.acquire(task.promise);
}
@ -191,14 +191,14 @@ public final class FixedChannelPool extends SimpleChannelPool {
assert executor.inEventLoop();
if (acquiredChannelCount < maxConnections) {
++acquiredChannelCount;
assert acquiredChannelCount > 0;
assert acquiredChannelCount >= 0;
// We need to create a new promise as we need to ensure the AcquireListener runs in the correct
// EventLoop
Promise<Channel> p = executor.newPromise();
p.addListener(new AcquireListener(promise));
AcquireListener l = new AcquireListener(promise);
l.acquired();
p.addListener(l);
super.acquire(p);
} else {
if (pendingAcquireCount >= maxPendingAcquires) {
@ -271,10 +271,8 @@ public final class FixedChannelPool extends SimpleChannelPool {
timeoutFuture.cancel(false);
}
task.acquired();
--pendingAcquireCount;
++acquiredChannelCount;
task.acquired();
super.acquire(task.promise);
}
@ -291,7 +289,7 @@ public final class FixedChannelPool extends SimpleChannelPool {
ScheduledFuture<?> timeoutFuture;
public AcquireTask(Promise<Channel> promise) {
super(promise, false);
super(promise);
// We need to create a new promise as we need to ensure the AcquireListener runs in the correct
// EventLoop.
this.promise = executor.<Channel>newPromise().addListener(this);
@ -327,12 +325,7 @@ public final class FixedChannelPool extends SimpleChannelPool {
protected boolean acquired;
AcquireListener(Promise<Channel> originalPromise) {
this(originalPromise, true);
}
protected AcquireListener(Promise<Channel> originalPromise, boolean acquired) {
this.originalPromise = originalPromise;
this.acquired = acquired;
}
@Override
@ -353,6 +346,10 @@ public final class FixedChannelPool extends SimpleChannelPool {
}
public void acquired() {
if (acquired) {
return;
}
acquiredChannelCount++;
acquired = true;
}
}