1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.eclipse.jetty.client;
20
21 import java.io.Closeable;
22 import java.io.IOException;
23 import java.util.ArrayList;
24 import java.util.List;
25 import java.util.concurrent.BlockingDeque;
26 import java.util.concurrent.BlockingQueue;
27 import java.util.concurrent.LinkedBlockingDeque;
28 import java.util.concurrent.atomic.AtomicInteger;
29 import java.util.concurrent.locks.ReentrantLock;
30
31 import org.eclipse.jetty.client.api.Connection;
32 import org.eclipse.jetty.client.api.Destination;
33 import org.eclipse.jetty.util.BlockingArrayQueue;
34 import org.eclipse.jetty.util.Promise;
35 import org.eclipse.jetty.util.component.ContainerLifeCycle;
36 import org.eclipse.jetty.util.component.Dumpable;
37 import org.eclipse.jetty.util.log.Log;
38 import org.eclipse.jetty.util.log.Logger;
39 import org.eclipse.jetty.util.thread.Sweeper;
40
41 public class ConnectionPool implements Closeable, Dumpable, Sweeper.Sweepable
42 {
43 protected static final Logger LOG = Log.getLogger(ConnectionPool.class);
44
45 private final AtomicInteger connectionCount = new AtomicInteger();
46 private final ReentrantLock lock = new ReentrantLock();
47 private final Destination destination;
48 private final int maxConnections;
49 private final Promise<Connection> requester;
50 private final BlockingDeque<Connection> idleConnections;
51 private final BlockingQueue<Connection> activeConnections;
52
53 public ConnectionPool(Destination destination, int maxConnections, Promise<Connection> requester)
54 {
55 this.destination = destination;
56 this.maxConnections = maxConnections;
57 this.requester = requester;
58 this.idleConnections = new LinkedBlockingDeque<>(maxConnections);
59 this.activeConnections = new BlockingArrayQueue<>(maxConnections);
60 }
61
62 public int getConnectionCount()
63 {
64 return connectionCount.get();
65 }
66
67 public BlockingQueue<Connection> getIdleConnections()
68 {
69 return idleConnections;
70 }
71
72 public BlockingQueue<Connection> getActiveConnections()
73 {
74 return activeConnections;
75 }
76
77 public Connection acquire()
78 {
79 Connection connection = activateIdle();
80 if (connection == null)
81 connection = tryCreate();
82 return connection;
83 }
84
85 private Connection tryCreate()
86 {
87 while (true)
88 {
89 int current = getConnectionCount();
90 final int next = current + 1;
91
92 if (next > maxConnections)
93 {
94 if (LOG.isDebugEnabled())
95 LOG.debug("Max connections {}/{} reached", current, maxConnections);
96
97 return activateIdle();
98 }
99
100 if (connectionCount.compareAndSet(current, next))
101 {
102 if (LOG.isDebugEnabled())
103 LOG.debug("Connection {}/{} creation", next, maxConnections);
104
105 destination.newConnection(new Promise<Connection>()
106 {
107 @Override
108 public void succeeded(Connection connection)
109 {
110 if (LOG.isDebugEnabled())
111 LOG.debug("Connection {}/{} creation succeeded {}", next, maxConnections, connection);
112
113 idleCreated(connection);
114
115 requester.succeeded(connection);
116 }
117
118 @Override
119 public void failed(Throwable x)
120 {
121 if (LOG.isDebugEnabled())
122 LOG.debug("Connection " + next + "/" + maxConnections + " creation failed", x);
123
124 connectionCount.decrementAndGet();
125
126 requester.failed(x);
127 }
128 });
129
130
131 return activateIdle();
132 }
133 }
134 }
135
136 protected void idleCreated(Connection connection)
137 {
138 boolean idle;
139 final ReentrantLock lock = this.lock;
140 lock.lock();
141 try
142 {
143
144 idle = idleConnections.offerLast(connection);
145 }
146 finally
147 {
148 lock.unlock();
149 }
150
151 idle(connection, idle);
152 }
153
154 private Connection activateIdle()
155 {
156 boolean acquired;
157 Connection connection;
158 final ReentrantLock lock = this.lock;
159 lock.lock();
160 try
161 {
162 connection = idleConnections.pollFirst();
163 if (connection == null)
164 return null;
165 acquired = activeConnections.offer(connection);
166 }
167 finally
168 {
169 lock.unlock();
170 }
171
172 if (acquired)
173 {
174 if (LOG.isDebugEnabled())
175 LOG.debug("Connection active {}", connection);
176 acquired(connection);
177 return connection;
178 }
179 else
180 {
181 if (LOG.isDebugEnabled())
182 LOG.debug("Connection active overflow {}", connection);
183 connection.close();
184 return null;
185 }
186 }
187
188 protected void acquired(Connection connection)
189 {
190 }
191
192 public boolean release(Connection connection)
193 {
194 boolean idle;
195 final ReentrantLock lock = this.lock;
196 lock.lock();
197 try
198 {
199 if (!activeConnections.remove(connection))
200 return false;
201
202 idle = idleConnections.offerFirst(connection);
203 }
204 finally
205 {
206 lock.unlock();
207 }
208
209 released(connection);
210 return idle(connection, idle);
211 }
212
213 protected boolean idle(Connection connection, boolean idle)
214 {
215 if (idle)
216 {
217 if (LOG.isDebugEnabled())
218 LOG.debug("Connection idle {}", connection);
219 return true;
220 }
221 else
222 {
223 if (LOG.isDebugEnabled())
224 LOG.debug("Connection idle overflow {}", connection);
225 connection.close();
226 return false;
227 }
228 }
229
230 protected void released(Connection connection)
231 {
232 }
233
234 public boolean remove(Connection connection)
235 {
236 boolean activeRemoved;
237 boolean idleRemoved;
238 final ReentrantLock lock = this.lock;
239 lock.lock();
240 try
241 {
242 activeRemoved = activeConnections.remove(connection);
243 idleRemoved = idleConnections.remove(connection);
244 }
245 finally
246 {
247 lock.unlock();
248 }
249
250 if (activeRemoved)
251 released(connection);
252 boolean removed = activeRemoved || idleRemoved;
253 if (removed)
254 {
255 int pooled = connectionCount.decrementAndGet();
256 if (LOG.isDebugEnabled())
257 LOG.debug("Connection removed {} - pooled: {}", connection, pooled);
258 }
259 return removed;
260 }
261
262 public boolean isActive(Connection connection)
263 {
264 final ReentrantLock lock = this.lock;
265 lock.lock();
266 try
267 {
268 return activeConnections.contains(connection);
269 }
270 finally
271 {
272 lock.unlock();
273 }
274 }
275
276 public boolean isIdle(Connection connection)
277 {
278 final ReentrantLock lock = this.lock;
279 lock.lock();
280 try
281 {
282 return idleConnections.contains(connection);
283 }
284 finally
285 {
286 lock.unlock();
287 }
288 }
289
290 public boolean isEmpty()
291 {
292 return connectionCount.get() == 0;
293 }
294
295 public void close()
296 {
297 List<Connection> idles = new ArrayList<>();
298 List<Connection> actives = new ArrayList<>();
299 final ReentrantLock lock = this.lock;
300 lock.lock();
301 try
302 {
303 idles.addAll(idleConnections);
304 idleConnections.clear();
305 actives.addAll(activeConnections);
306 activeConnections.clear();
307 }
308 finally
309 {
310 lock.unlock();
311 }
312
313 connectionCount.set(0);
314
315 for (Connection connection : idles)
316 connection.close();
317
318
319 for (Connection connection : actives)
320 connection.close();
321 }
322
323 @Override
324 public String dump()
325 {
326 return ContainerLifeCycle.dump(this);
327 }
328
329 @Override
330 public void dump(Appendable out, String indent) throws IOException
331 {
332 List<Connection> actives = new ArrayList<>();
333 List<Connection> idles = new ArrayList<>();
334 final ReentrantLock lock = this.lock;
335 lock.lock();
336 try
337 {
338 actives.addAll(activeConnections);
339 idles.addAll(idleConnections);
340 }
341 finally
342 {
343 lock.unlock();
344 }
345
346 ContainerLifeCycle.dumpObject(out, this);
347 ContainerLifeCycle.dump(out, indent, actives, idles);
348 }
349
350 @Override
351 public boolean sweep()
352 {
353 List<Sweeper.Sweepable> toSweep = new ArrayList<>();
354 final ReentrantLock lock = this.lock;
355 lock.lock();
356 try
357 {
358 for (Connection connection : getActiveConnections())
359 {
360 if (connection instanceof Sweeper.Sweepable)
361 toSweep.add(((Sweeper.Sweepable)connection));
362 }
363 }
364 finally
365 {
366 lock.unlock();
367 }
368
369 for (Sweeper.Sweepable candidate : toSweep)
370 {
371 if (candidate.sweep())
372 {
373 boolean removed = getActiveConnections().remove(candidate);
374 LOG.warn("Connection swept: {}{}{} from active connections{}{}",
375 candidate,
376 System.lineSeparator(),
377 removed ? "Removed" : "Not removed",
378 System.lineSeparator(),
379 dump());
380 }
381 }
382
383 return false;
384 }
385
386 @Override
387 public String toString()
388 {
389 int activeSize;
390 int idleSize;
391 final ReentrantLock lock = this.lock;
392 lock.lock();
393 try
394 {
395 activeSize = activeConnections.size();
396 idleSize = idleConnections.size();
397 }
398 finally
399 {
400 lock.unlock();
401 }
402
403 return String.format("%s[c=%d/%d,a=%d,i=%d]",
404 getClass().getSimpleName(),
405 connectionCount.get(),
406 maxConnections,
407 activeSize,
408 idleSize);
409 }
410 }