Skip to content

Commit 91ebeba

Browse files
slandelleslandelle
committed
cleanerPeriod should be half
1 parent f47fa7b commit 91ebeba

File tree

1 file changed

+65
-59
lines changed

1 file changed

+65
-59
lines changed

‎client/src/main/java/org/asynchttpclient/netty/channel/DefaultChannelPool.java

Lines changed: 65 additions & 59 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@
1313
*/
1414
package org.asynchttpclient.netty.channel;
1515

16-
import static org.asynchttpclient.util.Assertions.*;
16+
import static org.asynchttpclient.util.Assertions.assertNotNull;
1717
import static org.asynchttpclient.util.DateUtils.millisTime;
1818
import io.netty.channel.Channel;
1919
import io.netty.util.Timeout;
@@ -25,7 +25,7 @@
2525
import java.util.List;
2626
import java.util.Map;
2727
import java.util.concurrent.ConcurrentHashMap;
28-
import java.util.concurrent.ConcurrentLinkedQueue;
28+
import java.util.concurrent.ConcurrentLinkedDeque;
2929
import java.util.concurrent.TimeUnit;
3030
import java.util.concurrent.atomic.AtomicBoolean;
3131

@@ -37,22 +37,20 @@
3737
import org.slf4j.LoggerFactory;
3838

3939
/**
40-
* A simple implementation of
41-
* {@link ChannelPool} based on a
42-
* {@link java.util.concurrent.ConcurrentHashMap}
40+
* A simple implementation of {@link ChannelPool} based on a {@link java.util.concurrent.ConcurrentHashMap}
4341
*/
4442
public final class DefaultChannelPool implements ChannelPool {
4543

4644
private static final Logger LOGGER = LoggerFactory.getLogger(DefaultChannelPool.class);
4745

48-
private final ConcurrentHashMap<Object, ConcurrentLinkedQueue<IdleChannel>> partitions = new ConcurrentHashMap<>();
49-
private final ConcurrentHashMap<Integer, ChannelCreation> channelId2Creation = new ConcurrentHashMap<>();
46+
private final ConcurrentHashMap<Object, ConcurrentLinkedDeque<IdleChannel>> partitions = new ConcurrentHashMap<>();
47+
private final ConcurrentHashMap<Integer, ChannelCreation> channelId2Creation;
5048
private final AtomicBoolean isClosed = new AtomicBoolean(false);
5149
private final Timer nettyTimer;
5250
private final int maxConnectionTtl;
53-
private final boolean maxConnectionTtlDisabled;
54-
private final long maxIdleTime;
55-
private final boolean maxIdleTimeDisabled;
51+
private final boolean maxConnectionTtlEnabled;
52+
private final int maxIdleTime;
53+
private final boolean maxIdleTimeEnabled;
5654
private final long cleanerPeriod;
5755

5856
public DefaultChannelPool(AsyncHttpClientConfig config, Timer hashedWheelTimer) {
@@ -65,18 +63,24 @@ private int channelId(Channel channel) {
6563
return channel.hashCode();
6664
}
6765

68-
public DefaultChannelPool(long maxIdleTime,//
66+
private int cleanerPeriod(int ttl) {
67+
return (int) Math.ceil(ttl / 2.0);
68+
}
69+
70+
public DefaultChannelPool(int maxIdleTime,//
6971
int maxConnectionTtl,//
7072
Timer nettyTimer) {
71-
this.maxIdleTime = maxIdleTime;
73+
this.maxIdleTime = (int) maxIdleTime;
7274
this.maxConnectionTtl = maxConnectionTtl;
73-
maxConnectionTtlDisabled = maxConnectionTtl <= 0;
75+
maxConnectionTtlEnabled = maxConnectionTtl > 0;
76+
channelId2Creation = maxConnectionTtlEnabled ? new ConcurrentHashMap<>() : null;
7477
this.nettyTimer = nettyTimer;
75-
maxIdleTimeDisabled = maxIdleTime <= 0;
78+
maxIdleTimeEnabled = maxIdleTime > 0;
7679

77-
cleanerPeriod = Math.min(maxConnectionTtlDisabled ? Long.MAX_VALUE : maxConnectionTtl, maxIdleTimeDisabled ? Long.MAX_VALUE : maxIdleTime);
80+
// period is half
81+
cleanerPeriod = Math.min(maxConnectionTtlEnabled ? cleanerPeriod(maxConnectionTtl) : Integer.MAX_VALUE, maxIdleTimeEnabled ? cleanerPeriod(maxIdleTime) : Long.MAX_VALUE);
7882

79-
if (!maxConnectionTtlDisabled || !maxIdleTimeDisabled)
83+
if (maxConnectionTtlEnabled || maxIdleTimeEnabled)
8084
scheduleNewIdleChannelDetector(new IdleChannelDetector());
8185
}
8286

@@ -116,7 +120,7 @@ public int hashCode() {
116120
}
117121

118122
private boolean isTtlExpired(Channel channel, long now) {
119-
if (maxConnectionTtlDisabled)
123+
if (!maxConnectionTtlEnabled)
120124
return false;
121125

122126
ChannelCreation creation = channelId2Creation.get(channelId(channel));
@@ -130,14 +134,14 @@ private boolean isRemotelyClosed(Channel channel) {
130134
private final class IdleChannelDetector implements TimerTask {
131135

132136
private boolean isIdleTimeoutExpired(IdleChannel idleChannel, long now) {
133-
return !maxIdleTimeDisabled && now - idleChannel.start >= maxIdleTime;
137+
return maxIdleTimeEnabled && now - idleChannel.start >= maxIdleTime;
134138
}
135139

136-
private List<IdleChannel> expiredChannels(ConcurrentLinkedQueue<IdleChannel> partition, long now) {
140+
private List<IdleChannel> expiredChannels(ConcurrentLinkedDeque<IdleChannel> partition, long now) {
137141
// lazy create
138142
List<IdleChannel> idleTimeoutChannels = null;
139143
for (IdleChannel idleChannel : partition) {
140-
if (isTtlExpired(idleChannel.channel, now) || isIdleTimeoutExpired(idleChannel, now) || isRemotelyClosed(idleChannel.channel)) {
144+
if (isIdleTimeoutExpired(idleChannel, now) || isRemotelyClosed(idleChannel.channel) || isTtlExpired(idleChannel.channel, now)) {
141145
LOGGER.debug("Adding Candidate expired Channel {}", idleChannel.channel);
142146
if (idleTimeoutChannels == null)
143147
idleTimeoutChannels = new ArrayList<>();
@@ -153,7 +157,7 @@ private boolean isChannelCloseable(Channel channel) {
153157
if (attribute instanceof NettyResponseFuture) {
154158
NettyResponseFuture<?> future = (NettyResponseFuture<?>) attribute;
155159
if (!future.isDone()) {
156-
LOGGER.error("Future not in appropriate state %s, not closing", future);
160+
LOGGER.error("Future not in appropriate state {}, not closing", future);
157161
return false;
158162
}
159163
}
@@ -190,40 +194,38 @@ public void run(Timeout timeout) throws Exception {
190194
if (isClosed.get())
191195
return;
192196

193-
try {
194-
if (LOGGER.isDebugEnabled())
195-
for (Object key: partitions.keySet()) {
196-
LOGGER.debug("Entry count for : {} : {}", key, partitions.get(key).size());
197-
}
197+
if (LOGGER.isDebugEnabled())
198+
for (Object key : partitions.keySet()) {
199+
LOGGER.debug("Entry count for : {} : {}", key, partitions.get(key).size());
200+
}
198201

199-
long start = millisTime();
200-
int closedCount = 0;
201-
int totalCount = 0;
202+
long start = millisTime();
203+
int closedCount = 0;
204+
int totalCount = 0;
202205

203-
for (ConcurrentLinkedQueue<IdleChannel> partition : partitions.values()) {
206+
for (ConcurrentLinkedDeque<IdleChannel> partition : partitions.values()) {
204207

205-
// store in intermediate unsynchronized lists to minimize
206-
// the impact on the ConcurrentLinkedQueue
207-
if (LOGGER.isDebugEnabled())
208-
totalCount += partition.size();
208+
// store in intermediate unsynchronized lists to minimize
209+
// the impact on the ConcurrentLinkedDeque
210+
if (LOGGER.isDebugEnabled())
211+
totalCount += partition.size();
209212

210-
List<IdleChannel> closedChannels = closeChannels(expiredChannels(partition, start));
213+
List<IdleChannel> closedChannels = closeChannels(expiredChannels(partition, start));
211214

212-
if (!closedChannels.isEmpty()) {
215+
if (!closedChannels.isEmpty()) {
216+
if (maxConnectionTtlEnabled) {
213217
for (IdleChannel closedChannel : closedChannels)
214218
channelId2Creation.remove(channelId(closedChannel.channel));
215-
216-
partition.removeAll(closedChannels);
217-
closedCount += closedChannels.size();
218219
}
220+
221+
partition.removeAll(closedChannels);
222+
closedCount += closedChannels.size();
219223
}
224+
}
220225

226+
if (LOGGER.isDebugEnabled()) {
221227
long duration = millisTime() - start;
222-
223228
LOGGER.debug("Closed {} connections out of {} in {}ms", closedCount, totalCount, duration);
224-
225-
} catch (Throwable t) {
226-
LOGGER.error("uncaught exception!", t);
227229
}
228230

229231
scheduleNewIdleChannelDetector(timeout.task());
@@ -242,38 +244,38 @@ public boolean offer(Channel channel, Object partitionKey) {
242244
if (isTtlExpired(channel, now))
243245
return false;
244246

245-
boolean offered = offer0(channel, partitionKey,now);
246-
if (offered) {
247+
boolean offered = offer0(channel, partitionKey, now);
248+
if (maxConnectionTtlEnabled && offered) {
247249
registerChannelCreation(channel, partitionKey, now);
248250
}
249251

250252
return offered;
251253
}
252-
254+
253255
private boolean offer0(Channel channel, Object partitionKey, long now) {
254-
ConcurrentLinkedQueue<IdleChannel> partition = partitions.get(partitionKey);
256+
ConcurrentLinkedDeque<IdleChannel> partition = partitions.get(partitionKey);
255257
if (partition == null) {
256-
partition = partitions.computeIfAbsent(partitionKey, pk -> new ConcurrentLinkedQueue<>());
258+
partition = partitions.computeIfAbsent(partitionKey, pk -> new ConcurrentLinkedDeque<>());
257259
}
258-
return partition.add(new IdleChannel(channel, now));
260+
return partition.offerFirst(new IdleChannel(channel, now));
259261
}
260-
262+
261263
private void registerChannelCreation(Channel channel, Object partitionKey, long now) {
262264
if (channelId2Creation.containsKey(partitionKey)) {
263265
channelId2Creation.putIfAbsent(channelId(channel), new ChannelCreation(now, partitionKey));
264266
}
265267
}
266-
268+
267269
/**
268270
* {@inheritDoc}
269271
*/
270272
public Channel poll(Object partitionKey) {
271273

272274
IdleChannel idleChannel = null;
273-
ConcurrentLinkedQueue<IdleChannel> partition = partitions.get(partitionKey);
275+
ConcurrentLinkedDeque<IdleChannel> partition = partitions.get(partitionKey);
274276
if (partition != null) {
275277
while (idleChannel == null) {
276-
idleChannel = partition.poll();
278+
idleChannel = partition.pollFirst();
277279

278280
if (idleChannel == null)
279281
// pool is empty
@@ -291,7 +293,7 @@ else if (isRemotelyClosed(idleChannel.channel)) {
291293
* {@inheritDoc}
292294
*/
293295
public boolean removeAll(Channel channel) {
294-
ChannelCreation creation = channelId2Creation.remove(channelId(channel));
296+
ChannelCreation creation = maxConnectionTtlEnabled ? channelId2Creation.remove(channelId(channel)) : null;
295297
return !isClosed.get() && creation != null && partitions.get(creation.partitionKey).remove(channel);
296298
}
297299

@@ -309,23 +311,27 @@ public void destroy() {
309311
if (isClosed.getAndSet(true))
310312
return;
311313

312-
for (ConcurrentLinkedQueue<IdleChannel> partition : partitions.values()) {
314+
for (ConcurrentLinkedDeque<IdleChannel> partition : partitions.values()) {
313315
for (IdleChannel idleChannel : partition)
314316
close(idleChannel.channel);
315317
}
316318

317319
partitions.clear();
318-
channelId2Creation.clear();
320+
if (maxConnectionTtlEnabled) {
321+
channelId2Creation.clear();
322+
}
319323
}
320324

321325
private void close(Channel channel) {
322326
// FIXME pity to have to do this here
323327
Channels.setDiscard(channel);
324-
channelId2Creation.remove(channelId(channel));
328+
if (maxConnectionTtlEnabled) {
329+
channelId2Creation.remove(channelId(channel));
330+
}
325331
Channels.silentlyCloseChannel(channel);
326332
}
327333

328-
private void flushPartition(Object partitionKey, ConcurrentLinkedQueue<IdleChannel> partition) {
334+
private void flushPartition(Object partitionKey, ConcurrentLinkedDeque<IdleChannel> partition) {
329335
if (partition != null) {
330336
partitions.remove(partitionKey);
331337
for (IdleChannel idleChannel : partition)
@@ -341,7 +347,7 @@ public void flushPartition(Object partitionKey) {
341347
@Override
342348
public void flushPartitions(ChannelPoolPartitionSelector selector) {
343349

344-
for (Map.Entry<Object, ConcurrentLinkedQueue<IdleChannel>> partitionsEntry : partitions.entrySet()) {
350+
for (Map.Entry<Object, ConcurrentLinkedDeque<IdleChannel>> partitionsEntry : partitions.entrySet()) {
345351
Object partitionKey = partitionsEntry.getKey();
346352
if (selector.select(partitionKey))
347353
flushPartition(partitionKey, partitionsEntry.getValue());

0 commit comments

Comments
 (0)