Closed
Description
After we set .setMaxConnectionsPerHost(64), our server seemed to happily work. We can pound it with traffic and see very few issues with connections since its so efficient at pooling connections that are in good condition.
Lib: async-http-client 1.9.15
java version "1.7.0_67"
After a while, however (about 24 hours), we start getting the above exception coming from the ChannelManager.
Looking at the NettyResponseListener code, I noticed something odd.
In NettyRequestSender.sendRequestWithNewChannel(), there's this bit:
boolean channelPreempted = false;
String partition = null;
try { // Do not throw an exception when we need an extra connection for a
// redirect.
if (!reclaimCache) {
// only compute when maxConnectionPerHost is enabled
// FIXME clean up
if (config.getMaxConnectionsPerHost() > 0)
partition = future.getPartitionId();
channelManager.preemptChannel(partition);
}
if (asyncHandler instanceof AsyncHandlerExtensions)
AsyncHandlerExtensions.class.cast(asyncHandler).onOpenConnection();
ChannelFuture channelFuture = connect(request, uri, proxy, useProxy, bootstrap, asyncHandler);
channelFuture.addListener(new NettyConnectListener<T>(future, this, channelManager, channelPreempted, partition));
} catch (Throwable t) {
if (channelPreempted)
channelManager.abortChannelPreemption(partition);
abort(null, future, t.getCause() == null ? t : t.getCause());
}If you notice, channelPreempted never gets written to. Isn't channelPreempted = true missing from the block where the channel is preempted?
Shouldn't it be:
boolean channelPreempted = false;
String partition = null;
try { // Do not throw an exception when we need an extra connection for a
// redirect.
if (!reclaimCache) {
// only compute when maxConnectionPerHost is enabled
// FIXME clean up
if (config.getMaxConnectionsPerHost() > 0)
partition = future.getPartitionId();
channelManager.preemptChannel(partition);
channelPreempted = true;
}
if (asyncHandler instanceof AsyncHandlerExtensions)
AsyncHandlerExtensions.class.cast(asyncHandler).onOpenConnection();
ChannelFuture channelFuture = connect(request, uri, proxy, useProxy, bootstrap, asyncHandler);
channelFuture.addListener(new NettyConnectListener<T>(future, this, channelManager, channelPreempted, partition));
} catch (Throwable t) {
if (channelPreempted)
channelManager.abortChannelPreemption(partition);
abort(null, future, t.getCause() == null ? t : t.getCause());
}The same class for netty3 has the correct code: