Skip to content

Commit d542a61

Browse files
tsegismonttsegismont
committed
Fix memory leak with one-shot connection factories (#1311)
Fixes #1302 After creating an unpooled connection, add it as a close hook to the creating context. Also, in this case, make sure the connection factory is closed after usage. Signed-off-by: Thomas Segismont <tsegismont@gmail.com>
1 parent 769a6d2 commit d542a61

File tree

9 files changed

+116
-61
lines changed

9 files changed

+116
-61
lines changed

‎vertx-db2-client/src/main/java/io/vertx/db2client/impl/DB2ConnectionImpl.java‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -36,8 +36,7 @@ public static Future<DB2Connection> connect(Vertx vertx, DB2ConnectOptions optio
3636
} catch (Exception e) {
3737
return ctx.failedFuture(e);
3838
}
39-
ctx.addCloseHook(client);
40-
return (Future) client.connect(ctx, options);
39+
return prepareForClose(ctx, client.connect(ctx, options)).map(DB2Connection::cast);
4140
}
4241

4342
public DB2ConnectionImpl(ContextInternal context, ConnectionFactory factory, Connection conn) {

‎vertx-mssql-client/src/main/java/io/vertx/mssqlclient/impl/MSSQLConnectionImpl.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright (c) 2011-2019 Contributors to the Eclipse Foundation
2+
* Copyright (c) 2011-2023 Contributors to the Eclipse Foundation
33
*
44
* This program and the accompanying materials are made available under the
55
* terms of the Eclipse Public License 2.0 which is available at
@@ -35,8 +35,7 @@ public MSSQLConnectionImpl(ContextInternal context, ConnectionFactory factory, C
3535
public static Future<MSSQLConnection> connect(Vertx vertx, MSSQLConnectOptions options) {
3636
ContextInternal ctx = (ContextInternal) vertx.getOrCreateContext();
3737
MSSQLConnectionFactory client = new MSSQLConnectionFactory(ctx.owner());
38-
ctx.addCloseHook(client);
39-
return (Future)client.connect(ctx, options);
38+
return prepareForClose(ctx, client.connect(ctx, options)).map(MSSQLConnection::cast);
4039
}
4140

4241
@Override

‎vertx-mysql-client/src/main/java/io/vertx/mysqlclient/impl/MySQLConnectionImpl.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright (c) 2011-2020 Contributors to the Eclipse Foundation
2+
* Copyright (c) 2011-2023 Contributors to the Eclipse Foundation
33
*
44
* This program and the accompanying materials are made available under the
55
* terms of the Eclipse Public License 2.0 which is available at
@@ -36,8 +36,7 @@ public static Future<MySQLConnection> connect(ContextInternal ctx, MySQLConnectO
3636
} catch (Exception e) {
3737
return ctx.failedFuture(e);
3838
}
39-
ctx.addCloseHook(client);
40-
return (Future)client.connect(ctx, options);
39+
return prepareForClose(ctx, client.connect(ctx, options)).map(MySQLConnection::cast);
4140
}
4241

4342
public MySQLConnectionImpl(ContextInternal context, ConnectionFactory factory, Connection conn) {

‎vertx-oracle-client/src/main/java/io/vertx/oracleclient/impl/OracleConnectionImpl.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright (c) 2011-2022 Contributors to the Eclipse Foundation
2+
* Copyright (c) 2011-2023 Contributors to the Eclipse Foundation
33
*
44
* This program and the accompanying materials are made available under the
55
* terms of the Eclipse Public License 2.0 which is available at
@@ -30,7 +30,6 @@ public OracleConnectionImpl(ContextInternal context, ConnectionFactory factory,
3030
public static Future<OracleConnection> connect(Vertx vertx, OracleConnectOptions options) {
3131
ContextInternal ctx = (ContextInternal) vertx.getOrCreateContext();
3232
OracleConnectionFactory client = new OracleConnectionFactory(ctx.owner());
33-
ctx.addCloseHook(client);
34-
return (Future) client.connect(ctx, options);
33+
return prepareForClose(ctx, client.connect(ctx, options)).map(OracleConnection::cast);
3534
}
3635
}

‎vertx-pg-client/src/main/java/io/vertx/pgclient/impl/PgConnectionImpl.java‎

Lines changed: 3 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -16,24 +16,19 @@
1616
*/
1717
package io.vertx.pgclient.impl;
1818

19+
import io.vertx.core.*;
1920
import io.vertx.core.impl.ContextInternal;
20-
import io.vertx.core.spi.metrics.ClientMetrics;
2121
import io.vertx.pgclient.PgConnectOptions;
2222
import io.vertx.pgclient.PgConnection;
2323
import io.vertx.pgclient.PgNotice;
2424
import io.vertx.pgclient.PgNotification;
2525
import io.vertx.pgclient.impl.codec.NoticeResponse;
26+
import io.vertx.pgclient.impl.codec.TxFailedEvent;
2627
import io.vertx.pgclient.spi.PgDriver;
2728
import io.vertx.sqlclient.impl.Connection;
2829
import io.vertx.sqlclient.impl.Notification;
2930
import io.vertx.sqlclient.impl.SocketConnectionBase;
3031
import io.vertx.sqlclient.impl.SqlConnectionBase;
31-
import io.vertx.core.AsyncResult;
32-
import io.vertx.core.Context;
33-
import io.vertx.core.Future;
34-
import io.vertx.core.Handler;
35-
import io.vertx.core.Vertx;
36-
import io.vertx.pgclient.impl.codec.TxFailedEvent;
3732

3833
import java.util.function.Supplier;
3934

@@ -46,8 +41,7 @@ public static Future<PgConnection> connect(ContextInternal context, Supplier<PgC
4641
} catch (Exception e) {
4742
return context.failedFuture(e);
4843
}
49-
context.addCloseHook(client);
50-
return (Future) client.connect(context, options.get());
44+
return prepareForClose(context, client.connect(context, options.get())).map(PgConnection::cast);
5145
}
5246

5347
private volatile Handler<PgNotification> notificationHandler;

‎vertx-pg-client/src/test/java/io/vertx/pgclient/PgConnectionTestBase.java‎

Lines changed: 1 addition & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -17,18 +17,11 @@
1717

1818
package io.vertx.pgclient;
1919

20-
import io.vertx.core.AbstractVerticle;
2120
import io.vertx.core.AsyncResult;
22-
import io.vertx.core.Promise;
2321
import io.vertx.core.buffer.Buffer;
2422
import io.vertx.ext.unit.Async;
2523
import io.vertx.ext.unit.TestContext;
26-
import io.vertx.sqlclient.ProxyServer;
27-
import io.vertx.sqlclient.Row;
28-
import io.vertx.sqlclient.RowSet;
29-
import io.vertx.sqlclient.SqlConnection;
30-
import io.vertx.sqlclient.TransactionRollbackException;
31-
import io.vertx.sqlclient.Tuple;
24+
import io.vertx.sqlclient.*;
3225
import org.junit.Test;
3326

3427
import java.util.ArrayList;
@@ -319,24 +312,6 @@ public void testBatchInsertError(TestContext ctx) throws Exception {
319312
}));
320313
}
321314

322-
@Test
323-
public void testCloseOnUndeploy(TestContext ctx) {
324-
Async done = ctx.async();
325-
vertx.deployVerticle(new AbstractVerticle() {
326-
@Override
327-
public void start(Promise<Void> startPromise) throws Exception {
328-
connector.accept(ctx.asyncAssertSuccess(conn -> {
329-
conn.closeHandler(v -> {
330-
done.complete();
331-
});
332-
startPromise.complete();
333-
}));
334-
}
335-
}).onComplete(ctx.asyncAssertSuccess(id -> {
336-
vertx.undeploy(id);
337-
}));
338-
}
339-
340315
@Test
341316
public void testTransactionCommit(TestContext ctx) {
342317
testTransactionCommit(ctx, Runnable::run);

‎vertx-pg-client/src/test/java/io/vertx/pgclient/PgPooledConnectionTest.java‎

Lines changed: 1 addition & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -43,11 +43,7 @@ public void tearDown(TestContext ctx) {
4343
if (pool != null) {
4444
PgPool p = pool;
4545
pool = null;
46-
try {
47-
p.close();
48-
} catch (IllegalStateException e) {
49-
// Might be already closed because of testCloseOnUndeploy
50-
}
46+
p.close();
5147
}
5248
super.tearDown(ctx);
5349
}

‎vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/SqlConnectionBase.java‎

Lines changed: 27 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -17,15 +17,18 @@
1717

1818
package io.vertx.sqlclient.impl;
1919

20+
import io.vertx.core.*;
2021
import io.vertx.core.impl.ContextInternal;
2122
import io.vertx.core.impl.future.PromiseInternal;
2223
import io.vertx.core.spi.metrics.ClientMetrics;
2324
import io.vertx.core.spi.tracing.VertxTracer;
2425
import io.vertx.sqlclient.PrepareOptions;
2526
import io.vertx.sqlclient.PreparedStatement;
27+
import io.vertx.sqlclient.SqlConnection;
2628
import io.vertx.sqlclient.Transaction;
27-
import io.vertx.sqlclient.impl.command.*;
28-
import io.vertx.core.*;
29+
import io.vertx.sqlclient.impl.command.CommandBase;
30+
import io.vertx.sqlclient.impl.command.PrepareStatementCommand;
31+
import io.vertx.sqlclient.impl.command.QueryCommandBase;
2932
import io.vertx.sqlclient.impl.pool.SqlConnectionPool;
3033
import io.vertx.sqlclient.impl.tracing.QueryReporter;
3134
import io.vertx.sqlclient.spi.ConnectionFactory;
@@ -35,10 +38,11 @@
3538
/**
3639
* @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
3740
*/
38-
public class SqlConnectionBase<C extends SqlConnectionBase<C>> extends SqlClientBase implements SqlConnectionInternal {
41+
public class SqlConnectionBase<C extends SqlConnectionBase<C>> extends SqlClientBase implements SqlConnectionInternal, Closeable {
3942

4043
private volatile Handler<Throwable> exceptionHandler;
4144
private volatile Handler<Void> closeHandler;
45+
private volatile boolean closeFactoryAfterUsage;
4246
protected TransactionImpl tx;
4347
protected final ContextInternal context;
4448
protected final ConnectionFactory factory;
@@ -192,7 +196,16 @@ public Future<Void> close() {
192196
return promise.future();
193197
}
194198

195-
private void close(Promise<Void> promise) {
199+
@Override
200+
public void close(Promise<Void> completion) {
201+
doClose(completion);
202+
if (closeFactoryAfterUsage) {
203+
completion.future().onComplete(v -> factory.close(Promise.promise()));
204+
context.removeCloseHook(this);
205+
}
206+
}
207+
208+
private void doClose(Promise<Void> promise) {
196209
context.execute(promise, p -> {
197210
if (tx != null) {
198211
tx.rollback(ar -> conn.close(this, p));
@@ -202,4 +215,14 @@ private void close(Promise<Void> promise) {
202215
}
203216
});
204217
}
218+
219+
protected static Future<SqlConnection> prepareForClose(ContextInternal ctx, Future<SqlConnection> future) {
220+
return future.andThen(ar -> {
221+
if (ar.succeeded()) {
222+
SqlConnectionBase<?> base = (SqlConnectionBase<?>) ar.result();
223+
base.closeFactoryAfterUsage = true;
224+
ctx.addCloseHook(base);
225+
}
226+
});
227+
}
205228
}

‎vertx-sql-client/src/test/java/io/vertx/sqlclient/tck/ConnectionTestBase.java‎

Lines changed: 77 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright (c) 2011-2020 Contributors to the Eclipse Foundation
2+
* Copyright (c) 2011-2023 Contributors to the Eclipse Foundation
33
*
44
* This program and the accompanying materials are made available under the
55
* terms of the Eclipse Public License 2.0 which is available at
@@ -11,19 +11,24 @@
1111

1212
package io.vertx.sqlclient.tck;
1313

14-
import io.vertx.core.AsyncResult;
15-
import io.vertx.core.Handler;
16-
import io.vertx.core.Vertx;
14+
import io.vertx.core.*;
1715
import io.vertx.ext.unit.Async;
1816
import io.vertx.ext.unit.TestContext;
1917
import io.vertx.sqlclient.SqlConnectOptions;
2018
import io.vertx.sqlclient.SqlConnection;
19+
import io.vertx.sqlclient.impl.SqlConnectionBase;
20+
import io.vertx.sqlclient.spi.ConnectionFactory;
2121
import io.vertx.sqlclient.spi.DatabaseMetadata;
22-
2322
import org.junit.After;
2423
import org.junit.Before;
2524
import org.junit.Test;
2625

26+
import java.util.Collections;
27+
import java.util.Set;
28+
import java.util.WeakHashMap;
29+
30+
import static java.util.concurrent.TimeUnit.SECONDS;
31+
2732
public abstract class ConnectionTestBase {
2833
protected Vertx vertx;
2934
protected Connector<SqlConnection> connector;
@@ -46,7 +51,73 @@ public void tearDown(TestContext ctx) {
4651

4752
@Test
4853
public void testConnect(TestContext ctx) {
49-
connect(ctx.asyncAssertSuccess(conn -> {
54+
connect(ctx.asyncAssertSuccess());
55+
}
56+
57+
@Test
58+
public void testConnectNoLeak(TestContext ctx) throws Exception {
59+
Set<SqlConnectionBase<?>> connections = Collections.synchronizedSet(Collections.newSetFromMap(new WeakHashMap<>()));
60+
Set<ConnectionFactory> factories = Collections.synchronizedSet(Collections.newSetFromMap(new WeakHashMap<>()));
61+
Async async = ctx.async(100);
62+
for (int i = 0; i < 100; i++) {
63+
connect(ctx.asyncAssertSuccess(conn -> {
64+
SqlConnectionBase<?> base = (SqlConnectionBase<?>) conn;
65+
connections.add(base);
66+
factories.add(base.factory());
67+
conn.close().onComplete(ctx.asyncAssertSuccess(v -> async.countDown()));
68+
}));
69+
}
70+
async.awaitSuccess();
71+
for (int c = 0; c < 5; c++) {
72+
System.gc();
73+
SECONDS.sleep(1);
74+
}
75+
ctx.assertEquals(0, connections.size());
76+
ctx.assertEquals(0, factories.size());
77+
}
78+
79+
@Test
80+
public void testConnectNoLeakInVerticle(TestContext ctx) throws Exception {
81+
Set<SqlConnectionBase<?>> connections = Collections.synchronizedSet(Collections.newSetFromMap(new WeakHashMap<>()));
82+
Set<ConnectionFactory> factories = Collections.synchronizedSet(Collections.newSetFromMap(new WeakHashMap<>()));
83+
Async async = ctx.async(100);
84+
vertx.deployVerticle(new AbstractVerticle() {
85+
@Override
86+
public void start() throws Exception {
87+
for (int i = 0; i < 100; i++) {
88+
connect(ctx.asyncAssertSuccess(conn -> {
89+
SqlConnectionBase<?> base = (SqlConnectionBase<?>) conn;
90+
connections.add(base);
91+
factories.add(base.factory());
92+
conn.close().onComplete(ctx.asyncAssertSuccess(v -> async.countDown()));
93+
}));
94+
}
95+
}
96+
});
97+
async.awaitSuccess();
98+
for (int c = 0; c < 5; c++) {
99+
System.gc();
100+
SECONDS.sleep(1);
101+
}
102+
ctx.assertEquals(0, connections.size());
103+
ctx.assertEquals(0, factories.size());
104+
}
105+
106+
@Test
107+
public void testCloseOnUndeploy(TestContext ctx) {
108+
Async done = ctx.async();
109+
vertx.deployVerticle(new AbstractVerticle() {
110+
@Override
111+
public void start(Promise<Void> startPromise) throws Exception {
112+
connect(ctx.asyncAssertSuccess(conn -> {
113+
conn.closeHandler(v -> {
114+
done.complete();
115+
});
116+
startPromise.complete();
117+
}));
118+
}
119+
}).onComplete(ctx.asyncAssertSuccess(id -> {
120+
vertx.undeploy(id);
50121
}));
51122
}
52123

0 commit comments

Comments
 (0)