[GitHub] [iotdb] JackieTien97 commented on a change in pull request #2049: [IOTDB-1007]Fix session pool concurrency and leakage issue when pool.close is called

[GitBox <gi...@apache.org>]

JackieTien97 commented on a change in pull request #2049:
URL: https://github.com/apache/iotdb/pull/2049#discussion_r523859484



##########
File path: session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java
##########
@@ -76,6 +76,8 @@
   private boolean enableCompression = false;
   private ZoneId zoneId;
 
+  private boolean closed;//whether the queue is closed.

Review comment:
       I think it should be volatile, cause it is used outside a synchronized block somewhere below

##########
File path: session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java
##########
@@ -110,20 +112,64 @@ public SessionPool(String ip, int port, String user, String password, int maxSiz
   @SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
   private Session getSession() throws IoTDBConnectionException {
     Session session = queue.poll();
+    if (closed) {
+      throw new IoTDBConnectionException("Session pool is closed");
+    }
     if (session != null) {
       return session;
     } else {
+      long start = System.currentTimeMillis();
+      boolean canCreate = false;
       synchronized (this) {
-        long start = System.currentTimeMillis();
+        if (size < maxSize) {
+          //we can create more session
+          size++;
+          canCreate = true;
+          //but we do it after skip synchronized block because connection a session is time consuming.
+        }
+      }
+      if (canCreate) {
+        //create a new one.
+        if (logger.isDebugEnabled()) {
+          logger.debug("Create a new Session {}, {}, {}, {}", ip, port, user, password);
+        }
+        session = new Session(ip, port, user, password, fetchSize, zoneId);
+        try {
+          session.open(enableCompression);
+          //avoid someone has called close() the session pool
+          synchronized (this) {
+            if (closed) {
+              //have to release the connection...
+              session.close();
+              throw new IoTDBConnectionException("Session pool is closed");
+            } else {
+              return session;
+            }
+          }
+        } catch (IoTDBConnectionException e) {
+          //if exception, we will throw the exception.
+          //Meanwhile, we have to set size--
+          synchronized (this) {
+            size--;
+            this.notifyAll();

Review comment:
       I think notify() is better, cause only one thread can get session.

##########
File path: session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java
##########
@@ -223,10 +260,12 @@ public void closeResultSet(SessionDataSetWrapper wrapper) {
   }
 
   private synchronized void removeSession() {
+    logger.warn("Remove a broken Session {}, {}, {}", ip, port, user);
+    size--;
+    this.notifyAll();

Review comment:
       notify() is better.

##########
File path: session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java
##########
@@ -178,6 +207,10 @@ private void putBack(Session session) {
     queue.push(session);
     synchronized (this) {
       this.notifyAll();

Review comment:
       notify() is better.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org