Merge pull request #3864 from katzyn/loom

Improve performance of TCP client driver when it is used from virtual threads

h2/src/main/org/h2/command/Command.java

@@ -181,7 +181,8 @@ public ResultInterface executeQuery(long maxrows, boolean scrollable) {
         Database database = getDatabase();
         session.waitIfExclusiveModeEnabled();
         boolean callStop = true;
-        synchronized (session) {
+        session.lock();
+        try {
             session.startStatementWithinTransaction(this);
             Session oldSession = session.setThreadLocalSession();
             try {
@@ -230,14 +231,17 @@ public ResultInterface executeQuery(long maxrows, boolean scrollable) {
                     stop();
                 }
             }
+        } finally {
+            session.unlock();
         }
     }
 
     @Override
     public ResultWithGeneratedKeys executeUpdate(Object generatedKeysRequest) {
         long start = 0;
         boolean callStop = true;
-        synchronized (session) {
+        session.lock();
+        try {
             Database database = getDatabase();
             session.waitIfExclusiveModeEnabled();
             commitIfNonTransactional();
@@ -300,6 +304,8 @@ public ResultWithGeneratedKeys executeUpdate(Object generatedKeysRequest) {
                     }
                 }
             }
+        } finally {
+            session.unlock();
         }
     }
 

h2/src/main/org/h2/command/CommandRemote.java

@@ -120,7 +120,9 @@ private void prepareIfRequired() {
 
     @Override
     public ResultInterface getMetaData() {
-        synchronized (session) {
+        final SessionRemote session = this.session;
+        session.lock();
+        try {
             if (!isQuery) {
                 return null;
             }
@@ -144,13 +146,17 @@ public ResultInterface getMetaData() {
             }
             session.autoCommitIfCluster();
             return result;
+        } finally {
+            session.unlock();
         }
     }
 
     @Override
     public ResultInterface executeQuery(long maxRows, boolean scrollable) {
         checkParameters();
-        synchronized (session) {
+        final SessionRemote session = this.session;
+        session.lock();
+        try {
             int objectId = session.getNextId();
             ResultRemote result = null;
             for (int i = 0, count = 0; i < transferList.size(); i++) {
@@ -185,6 +191,8 @@ public ResultInterface executeQuery(long maxRows, boolean scrollable) {
             session.autoCommitIfCluster();
             session.readSessionState();
             return result;
+        } finally {
+            session.unlock();
         }
     }
 
@@ -194,7 +202,9 @@ public ResultWithGeneratedKeys executeUpdate(Object generatedKeysRequest) {
         int generatedKeysMode = GeneratedKeysMode.valueOf(generatedKeysRequest);
         boolean readGeneratedKeys = generatedKeysMode != GeneratedKeysMode.NONE;
         int objectId = readGeneratedKeys ? session.getNextId() : 0;
-        synchronized (session) {
+        final SessionRemote session = this.session;
+        session.lock();
+        try {
             long updateCount = 0L;
             ResultRemote generatedKeys = null;
             boolean autoCommit = false;
@@ -246,6 +256,8 @@ public ResultWithGeneratedKeys executeUpdate(Object generatedKeysRequest) {
                 return new ResultWithGeneratedKeys.WithKeys(updateCount, generatedKeys);
             }
             return ResultWithGeneratedKeys.of(updateCount);
+        } finally {
+            session.unlock();
         }
     }
 
@@ -273,10 +285,12 @@ private void sendParameters(Transfer transfer) throws IOException {
 
     @Override
     public void close() {
+        final SessionRemote session = this.session;
         if (session == null || session.isClosed()) {
             return;
         }
-        synchronized (session) {
+        session.lock();
+        try {
             session.traceOperation("COMMAND_CLOSE", id);
             for (Transfer transfer : transferList) {
                 try {
@@ -285,8 +299,10 @@ public void close() {
                     trace.error(e, "close");
                 }
             }
+        } finally {
+            session.unlock();
         }
-        session = null;
+        this.session = null;
         try {
             for (ParameterInterface p : parameters) {
                 Value v = p.getParamValue();

h2/src/main/org/h2/command/Parser.java

@@ -7088,7 +7088,8 @@ private TableView createCTEView(String cteViewName, String querySQL, ArrayList<P
         // it twice - once without the flag set, and if we didn't see a recursive term,
         // then we just compile it again.
         TableView view;
-        synchronized (session) {
+        session.lock();
+        try {
             view = new TableView(schema, id, cteViewName, querySQL,
                     queryParameters, columnTemplateArray, session,
                     true, false, true,
@@ -7110,6 +7111,8 @@ private TableView createCTEView(String cteViewName, String querySQL, ArrayList<P
             }
             // both removeSchemaObject and removeLocalTempTable hold meta locks
             database.unlockMeta(session);
+        } finally {
+            session.unlock();
         }
         view.setTableExpression(true);
         view.setTemporary(isTemporary);

h2/src/main/org/h2/engine/Database.java

@@ -607,7 +607,9 @@ private void executeMeta() {
                 lastRecords.add(rec);
             }
         }
-        synchronized (systemSession) {
+        final SessionLocal systemSession = this.systemSession;
+        systemSession.lock();
+        try {
             executeMeta(firstRecords);
             // Domains may depend on other domains
             int count = domainRecords.size();
@@ -652,6 +654,8 @@ private void executeMeta() {
                 }
             }
             executeMeta(lastRecords);
+        } finally {
+            systemSession.unlock();
         }
     }
 

h2/src/main/org/h2/engine/Engine.java

@@ -235,7 +235,8 @@ private static SessionLocal openSession(ConnectionInfo ci) {
                 throw DbException.get(ErrorCode.DATABASE_CALLED_AT_SHUTDOWN);
             }
         }
-        synchronized (session) {
+        session.lock();
+        try {
             session.setAllowLiterals(true);
             DbSettings defaultSettings = DbSettings.DEFAULT;
             for (String setting : ci.getKeys()) {
@@ -286,6 +287,8 @@ private static SessionLocal openSession(ConnectionInfo ci) {
             }
             session.setAllowLiterals(false);
             session.commit(true);
+        } finally {
+            session.unlock();
         }
         return session;
     }

h2/src/main/org/h2/engine/Session.java

@@ -6,6 +6,7 @@
 package org.h2.engine;
 
 import java.util.ArrayList;
+import java.util.concurrent.locks.ReentrantLock;
 
 import org.h2.command.CommandInterface;
 import org.h2.jdbc.meta.DatabaseMeta;
@@ -90,6 +91,8 @@ public DynamicSettings(Mode mode, TimeZoneProvider timeZone) {
 
     }
 
+    private final ReentrantLock lock = new ReentrantLock();
+
     private ArrayList<String> sessionState;
 
     boolean sessionStateChanged;
@@ -101,6 +104,32 @@ public DynamicSettings(Mode mode, TimeZoneProvider timeZone) {
     Session() {
     }
 
+    /**
+     * Locks this session with a reentrant lock.
+     *
+     * <pre>
+     * final Session session = ...;
+     * session.lock();
+     * try {
+     *     ...
+     * } finally {
+     *     session.unlock();
+     * }
+     * </pre>
+     */
+    public final void lock() {
+        lock.lock();
+    }
+
+    /**
+     * Unlocks this session.
+     *
+     * @see #lock()
+     */
+    public final void unlock() {
+        lock.unlock();
+    }
+
     /**
      * Get the list of the cluster servers for this session.
      *

h2/src/main/org/h2/engine/SessionLocal.java

@@ -553,9 +553,13 @@ public void setLockTimeout(int lockTimeout) {
     }
 
     @Override
-    public synchronized CommandInterface prepareCommand(String sql,
-            int fetchSize) {
-        return prepareLocal(sql);
+    public CommandInterface prepareCommand(String sql, int fetchSize) {
+        lock();
+        try {
+            return prepareLocal(sql);
+        } finally {
+            unlock();
+        }
     }
 
     /**