Correctly count acquired channels when timeout occurs in FixedChannelPool
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:
parent
8b3fef96cc
commit
dbc078ccf4
@ -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;
|
||||
}
|
||||
}
|
||||
|
Loading…
Reference in New Issue
Block a user