Skip to content

Commit fcd8493

Browse files
rbygraverbygrave
authored
Merge pull request #134 from ebean-orm/feature/close-async-2
Close connections asynchronously (2)
2 parents eea5673 + 5d07893 commit fcd8493

10 files changed

Lines changed: 475 additions & 47 deletions

File tree

‎.github/workflows/build.yml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@ jobs:
1313
strategy:
1414
fail-fast: false
1515
matrix:
16-
java_version: [11]
16+
java_version: [21]
1717
os: [ubuntu-latest]
1818

1919
steps:

‎ebean-datasource/pom.xml‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -83,6 +83,67 @@
8383
<scope>test</scope>
8484
</dependency>
8585

86+
<dependency>
87+
<groupId>org.mariadb.jdbc</groupId>
88+
<artifactId>mariadb-java-client</artifactId>
89+
<version>3.5.3</version>
90+
<scope>test</scope>
91+
</dependency>
92+
8693
</dependencies>
8794

95+
<build>
96+
<plugins>
97+
98+
<plugin> <!-- Multi-Release with 21 -->
99+
<groupId>org.apache.maven.plugins</groupId>
100+
<artifactId>maven-jar-plugin</artifactId>
101+
<version>3.3.0</version>
102+
<configuration>
103+
<archive>
104+
<manifestEntries>
105+
<Multi-Release>true</Multi-Release>
106+
</manifestEntries>
107+
</archive>
108+
</configuration>
109+
</plugin>
110+
111+
<plugin>
112+
<groupId>org.apache.maven.plugins</groupId>
113+
<artifactId>maven-compiler-plugin</artifactId>
114+
<version>3.14.0</version>
115+
<executions>
116+
<!-- Compile for base version Java 11 -->
117+
<execution>
118+
<id>base</id>
119+
<goals>
120+
<goal>compile</goal>
121+
</goals>
122+
<configuration>
123+
<release>11</release>
124+
<compileSourceRoots>
125+
<compileSourceRoot>${project.basedir}/src/main/java</compileSourceRoot>
126+
</compileSourceRoots>
127+
</configuration>
128+
</execution>
129+
<!-- Compile for Java 21 -->
130+
<execution>
131+
<id>java21</id>
132+
<goals>
133+
<goal>compile</goal>
134+
</goals>
135+
<configuration>
136+
<release>21</release>
137+
<compileSourceRoots>
138+
<compileSourceRoot>${project.basedir}/src/main/java21</compileSourceRoot>
139+
</compileSourceRoots>
140+
<outputDirectory>${project.build.outputDirectory}/META-INF/versions/21</outputDirectory>
141+
</configuration>
142+
</execution>
143+
</executions>
144+
</plugin>
145+
146+
</plugins>
147+
</build>
148+
88149
</project>

‎ebean-datasource/src/main/java/io/ebean/datasource/pool/ConnectionPool.java‎

Lines changed: 78 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,9 @@
66
import java.io.PrintWriter;
77
import java.sql.*;
88
import java.util.*;
9+
import java.util.concurrent.ExecutorService;
10+
import java.util.concurrent.RejectedExecutionException;
11+
import java.util.concurrent.TimeUnit;
912
import java.util.concurrent.atomic.AtomicBoolean;
1013
import java.util.concurrent.atomic.AtomicInteger;
1114
import java.util.concurrent.atomic.LongAdder;
@@ -81,6 +84,8 @@ final class ConnectionPool implements DataSourcePool {
8184
private final PooledConnectionQueue queue;
8285
private Timer heartBeatTimer;
8386
private int heartbeatPoolExhaustedCount;
87+
private final ExecutorService executor;
88+
8489
/**
8590
* Used to find and close() leaked connections. Leaked connections are
8691
* thought to be busy but have not been used for some time. Each time a
@@ -134,6 +139,7 @@ final class ConnectionPool implements DataSourcePool {
134139
init();
135140
}
136141
this.nextTrimTime = System.currentTimeMillis() + trimPoolFreqMillis;
142+
this.executor = ExecutorFactory.newExecutor();
137143
}
138144

139145
private void init() {
@@ -649,15 +655,23 @@ public void offline() {
649655
shutdownPool(false, false);
650656
}
651657

652-
private void shutdownPool(boolean closeBusyConnections, boolean fromHook) {
653-
stopHeartBeatIfRunning();
654-
PoolStatus status = queue.shutdown(closeBusyConnections);
655-
dataSourceUp.set(false);
656-
if (fromHook) {
657-
Log.info("DataSource [{0}] shutdown on JVM exit {1} psc[hit:{2} miss:{3} put:{4} rem:{5}]", name, status, pscHit, pscMiss, pscPut, pscRem);
658-
} else {
659-
Log.info("DataSource [{0}] shutdown {1} psc[hit:{2} miss:{3} put:{4} rem:{5}]", name, status, pscHit, pscMiss, pscPut, pscRem);
660-
removeShutdownHook();
658+
private void shutdownPool(boolean fullShutdown, boolean fromHook) {
659+
heartbeatLock.lock();
660+
try {
661+
stopHeartBeatIfRunning();
662+
PoolStatus status = queue.shutdown(fullShutdown);
663+
dataSourceUp.set(false);
664+
if (fullShutdown) {
665+
shutdownExecutor();
666+
}
667+
if (fromHook) {
668+
Log.info("DataSource [{0}] shutdown on JVM exit {1} psc[hit:{2} miss:{3} put:{4} rem:{5}]", name, status, pscHit, pscMiss, pscPut, pscRem);
669+
} else {
670+
Log.info("DataSource [{0}] shutdown {1} psc[hit:{2} miss:{3} put:{4} rem:{5}]", name, status, pscHit, pscMiss, pscPut, pscRem);
671+
removeShutdownHook();
672+
}
673+
} finally {
674+
heartbeatLock.unlock();
661675
}
662676
}
663677

@@ -718,6 +732,61 @@ private void stopHeartBeatIfRunning() {
718732
}
719733
}
720734

735+
private static final class AsyncCloser implements Runnable {
736+
final PooledConnection pc;
737+
final boolean logErrors;
738+
739+
private AsyncCloser(PooledConnection pc, boolean logErrors) {
740+
this.pc = pc;
741+
this.logErrors = logErrors;
742+
}
743+
744+
@Override
745+
public void run() {
746+
pc.doCloseConnection(logErrors);
747+
}
748+
749+
@Override
750+
public String toString() {
751+
return pc.toString();
752+
}
753+
}
754+
755+
/**
756+
* Closes the connection in the background as it may be slow or block.
757+
*/
758+
void closeConnectionFullyAsync(PooledConnection pc, boolean logErrors) {
759+
if (!executor.isShutdown()) {
760+
try {
761+
executor.submit(new AsyncCloser(pc, logErrors));
762+
return;
763+
} catch (RejectedExecutionException e) {
764+
Log.trace("DataSource [{0}] closing connection synchronously", name);
765+
}
766+
}
767+
// it is possible that we receive runnables after shutdown.
768+
// in this case, we will execute them immediately (outside lock)
769+
pc.doCloseConnection(logErrors);
770+
}
771+
772+
private void shutdownExecutor() {
773+
executor.shutdown();
774+
try {
775+
if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
776+
Log.warn("DataSource [{0}] on shutdown, timeout waiting for connections to close", name);
777+
}
778+
} catch (InterruptedException ie) {
779+
Log.warn("DataSource [{0}] on shutdown, interrupted closing connections", name, ie);
780+
}
781+
final var pendingTasks = executor.shutdownNow();
782+
if (!pendingTasks.isEmpty()) {
783+
Log.warn("DataSource [{0}] on shutdown, {1} pending connections were not closed", name, pendingTasks.size());
784+
}
785+
}
786+
787+
/**
788+
* Return the default autoCommit setting for the pool.
789+
*/
721790
@Override
722791
public boolean isAutoCommit() {
723792
return autoCommit;

‎ebean-datasource/src/main/java/io/ebean/datasource/pool/ExecutorFactory.java‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
package io.ebean.datasource.pool;
2+
3+
import java.util.concurrent.ExecutorService;
4+
import java.util.concurrent.Executors;
5+
6+
final class ExecutorFactory {
7+
8+
static ExecutorService newExecutor() {
9+
return Executors.newSingleThreadExecutor();
10+
}
11+
}

‎ebean-datasource/src/main/java/io/ebean/datasource/pool/PooledConnection.java‎

Lines changed: 57 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
import java.sql.*;
44
import java.util.ArrayList;
55
import java.util.Map;
6+
import java.util.concurrent.TimeUnit;
67
import java.util.concurrent.locks.ReentrantLock;
78

89
/**
@@ -234,52 +235,75 @@ void closeConnectionFully(boolean logErrors) {
234235
if (Log.isLoggable(System.Logger.Level.TRACE)) {
235236
Log.trace("Closing Connection[{0}] reason[{1}], pstmtStats: {2}", name, closeReason, pstmtCache.description());
236237
}
237-
if (pool != null) {
238-
pool.pstmtCacheMetrics(pstmtCache);
238+
if (pool == null) {
239+
return; // this can happen in tests only.
239240
}
241+
pool.pstmtCacheMetrics(pstmtCache);
242+
pool.closeConnectionFullyAsync(this, logErrors);
243+
}
244+
245+
/**
246+
* this method performs network IO and may block
247+
*/
248+
void doCloseConnection(boolean logErrors) {
249+
pool.dec();
250+
long start = System.nanoTime();
240251
try {
241-
if (connection.isClosed()) {
242-
// Typically, the JDBC Driver has its own JVM shutdown hook and already
243-
// closed the connections in our DataSource pool so making this DEBUG level
244-
Log.trace("Closing Connection[{0}] that is already closed?", name);
245-
return;
252+
try {
253+
if (connection.isClosed()) {
254+
// Typically, the JDBC Driver has its own JVM shutdown hook and already
255+
// closed the connections in our DataSource pool so making this DEBUG level
256+
Log.trace("Closing Connection[{0}] that is already closed?", name);
257+
return;
258+
}
259+
} catch (SQLException ex) {
260+
if (logErrors) {
261+
Log.error("Error checking if connection [" + name + "] is closed", ex);
262+
}
246263
}
247-
} catch (SQLException ex) {
248-
if (logErrors) {
249-
Log.error("Error checking if connection [" + name + "] is closed", ex);
264+
try {
265+
clearPreparedStatementCache();
266+
} catch (SQLException ex) {
267+
if (logErrors) {
268+
Log.warn("Error when closing connection Statements", ex);
269+
}
270+
}
271+
try {
272+
// DB2 (and some other DBMS) may have uncommitted changes and do not allow close
273+
// so try to do a rollback.
274+
if (!connection.getAutoCommit()) {
275+
connection.rollback();
276+
}
277+
} catch (SQLException ex) {
278+
if (logErrors) {
279+
Log.warn("Could not perform rollback", ex);
280+
}
281+
}
282+
try {
283+
connection.close();
284+
} catch (SQLException ex) {
285+
if (logErrors || Log.isLoggable(System.Logger.Level.DEBUG)) {
286+
Log.error("Error when fully closing connection [" + fullDescription() + "]", ex);
287+
}
288+
}
289+
} finally {
290+
long millis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
291+
if (millis > 500) {
292+
Log.warn("Closing connection [" + fullDescription() + "] took an unexpected long time of " + millis + " ms");
250293
}
251294
}
295+
}
296+
297+
void clearPreparedStatementCache() throws SQLException {
252298
lock.lock();
253299
try {
254300
for (ExtendedPreparedStatement ps : pstmtCache.values()) {
255301
ps.closeDestroy();
256302
}
257-
} catch (SQLException ex) {
258-
if (logErrors) {
259-
Log.warn("Error when closing connection Statements", ex);
260-
}
303+
261304
} finally {
262305
lock.unlock();
263306
}
264-
try {
265-
// DB2 (and some other DBMS) may have uncommitted changes and do not allow close
266-
// so try to do a rollback.
267-
if (!connection.getAutoCommit()) {
268-
connection.rollback();
269-
}
270-
} catch (SQLException ex) {
271-
if (logErrors) {
272-
Log.warn("Could not perform rollback", ex);
273-
}
274-
}
275-
try {
276-
connection.close();
277-
pool.dec();
278-
} catch (SQLException ex) {
279-
if (logErrors || Log.isLoggable(System.Logger.Level.DEBUG)) {
280-
Log.error("Error when fully closing connection [" + fullDescription() + "]", ex);
281-
}
282-
}
283307
}
284308

285309
/**

‎ebean-datasource/src/main/java21/io/ebean/datasource/pool/ExecutorFactory.java‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
package io.ebean.datasource.pool;
2+
3+
import java.util.concurrent.ExecutorService;
4+
import java.util.concurrent.Executors;
5+
6+
final class ExecutorFactory {
7+
8+
static ExecutorService newExecutor() {
9+
return Executors.newVirtualThreadPerTaskExecutor();
10+
}
11+
}

0 commit comments

Comments
 (0)