Add management of the lifecycle of all Sessions (#3483)

Author: liujiajun

docs/UserGuide/IoTDB-SQL-Language/Maintenance-Command.md

@@ -59,10 +59,15 @@ IoTDB> CREATE SNAPSHOT FOR SCHEMA
 ```
 
 
-## Kill Query
+## Timeout
 
-When using IoTDB, you may encounter the following situations: you have entered a query statement, but can not get the result for a long time, as this query contains too much data or some other reasons, and have to wait until the query ends.
-Since version 0.12, IoTDB has provided two solutions for queries with long execution time: query timeout and query abort.
+IoTDB supports session and query level timeout.
+
+### Session timeout
+
+Session timeout controls when idle sessions are closed. An idle session is one that had not initiated any query or non-query operations for a period of time.
+
+Session timeout is disabled by default and can be set using the `session_timeout_threshold` parameter in IoTDB configuration file.
 
 ### Query timeout
 
@@ -73,7 +78,7 @@ IoTDB> select * from root;
 Msg: 701 Current query is time out, please check your statement or modify timeout parameter.
 ```
 
-The default timeout of the system is 60000 ms，which can be customized in the configuration file through the `query_timeout_threshold` parameter.
+The default timeout of a query is 60000 ms，which can be customized in the configuration file through the `query_timeout_threshold` parameter.
 
 If you use JDBC or Session, we also support setting a timeout for a single query（Unit: ms）：
 

docs/zh/UserGuide/IoTDB-SQL-Language/Maintenance-Command.md

@@ -56,11 +56,15 @@ IoTDB> CLEAR CACHE
 IoTDB> CREATE SNAPSHOT FOR SCHEMA
 ```
 
-## 中止查询
+## 超时
 
-当使用 IoTDB 时，您可能会遇到以下情形：输入了一个查询，但是由于其包含的数据量过大或是其他原因，导致长时间无法返回结果，但是迫于生产环境无法中止该命令，只能被迫等待。
+IoTDB 支持 Session 超时和查询超时。
 
-从 0.12 版本开始，IoTDB 对执行时间过长的查询给出了两种解决方案：查询超时和查询中止。
+### Session 超时
+
+Session 超时控制何时关闭空闲 Session。空闲 Session 指在一段时间内没有发起任何操作的 Session。
+
+Session 超时默认未开启。可以在配置文件中通过 `session_timeout_threshold` 参数进行配置。
 
 ### 查询超时
 

server/src/assembly/resources/conf/iotdb-engine.properties

@@ -462,6 +462,12 @@ timestamp_precision=ms
 # Datatype: int
 # merge_write_throughput_mb_per_sec=8
 
+# The maximum session idle time. unit: ms
+# Idle sessions are the ones that performs neither query or non-query operations for a period of time
+# Set to 0 to disable session timeout
+# Datatype: int
+# session_timeout_threshold=0
+
 # The max executing time of query. unit: ms
 # Datatype: int
 # query_timeout_threshold=60000

server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java

@@ -407,6 +407,9 @@ public class IoTDBConfig {
   /** the max executing time of query in ms. Unit: millisecond */
   private int queryTimeoutThreshold = 60000;
 
+  /** the max time to live of a session in ms. Unit: millisecond */
+  private int sessionTimeoutThreshold = 0;
+
   /** Replace implementation class of JDBC service */
   private String rpcImplClassName = TSServiceImpl.class.getName();
 
@@ -1217,6 +1220,14 @@ public void setQueryTimeoutThreshold(int queryTimeoutThreshold) {
     this.queryTimeoutThreshold = queryTimeoutThreshold;
   }
 
+  public int getSessionTimeoutThreshold() {
+    return sessionTimeoutThreshold;
+  }
+
+  public void setSessionTimeoutThreshold(int sessionTimeoutThreshold) {
+    this.sessionTimeoutThreshold = sessionTimeoutThreshold;
+  }
+
   public boolean isReadOnly() {
     return readOnly;
   }

server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java

@@ -360,6 +360,12 @@ private void loadProps() {
               properties.getProperty(
                   "query_timeout_threshold", Integer.toString(conf.getQueryTimeoutThreshold()))));
 
+      conf.setSessionTimeoutThreshold(
+          Integer.parseInt(
+              properties.getProperty(
+                  "session_timeout_threshold",
+                  Integer.toString(conf.getSessionTimeoutThreshold()))));
+
       conf.setSyncEnable(
           Boolean.parseBoolean(
               properties.getProperty("is_sync_enable", Boolean.toString(conf.isSyncEnable()))));

server/src/main/java/org/apache/iotdb/db/query/control/QueryTimeManager.java

@@ -94,6 +94,8 @@ public AtomicBoolean unRegisterQuery(long queryId) {
           if (scheduledFuture != null) {
             scheduledFuture.cancel(false);
           }
+          SessionTimeoutManager.getInstance()
+              .refresh(SessionManager.getInstance().getSessionIdByQueryId(queryId));
           return null;
         });
     return successRemoved;

server/src/main/java/org/apache/iotdb/db/query/control/SessionManager.java

@@ -77,6 +77,20 @@ public boolean releaseSessionResource(long sessionId) {
     return sessionIdToUsername.remove(sessionId) != null;
   }
 
+  public long getSessionIdByQueryId(long queryId) {
+    // TODO: make this more efficient with a queryId -> sessionId map
+    for (Map.Entry<Long, Set<Long>> statementToQueries : statementIdToQueryId.entrySet()) {
+      if (statementToQueries.getValue().contains(queryId)) {
+        for (Map.Entry<Long, Set<Long>> sessionToStatements : sessionIdToStatementId.entrySet()) {
+          if (sessionToStatements.getValue().contains(statementToQueries.getKey())) {
+            return sessionToStatements.getKey();
+          }
+        }
+      }
+    }
+    return -1;
+  }
+
   public long requestStatementId(long sessionId) {
     long statementId = statementIdGenerator.incrementAndGet();
     sessionIdToStatementId

server/src/main/java/org/apache/iotdb/db/query/control/SessionTimeoutManager.java

@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.db.query.control;
+
+import org.apache.iotdb.db.concurrent.IoTDBThreadPoolFactory;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+public class SessionTimeoutManager {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(SessionTimeoutManager.class);
+  private static final long MINIMUM_CLEANUP_PERIOD = 2000;
+  private static final long SESSION_TIMEOUT =
+      IoTDBDescriptor.getInstance().getConfig().getSessionTimeoutThreshold();
+
+  private Map<Long, Long> sessionIdToLastActiveTime;
+  private ScheduledExecutorService executorService;
+
+  private SessionTimeoutManager() {
+    if (SESSION_TIMEOUT == 0) {
+      return;
+    }
+
+    this.sessionIdToLastActiveTime = new ConcurrentHashMap<>();
+    this.executorService =
+        IoTDBThreadPoolFactory.newScheduledThreadPool(1, "session-timeout-manager");
+
+    executorService.scheduleAtFixedRate(
+        () -> {
+          LOGGER.info("cleaning up expired sessions");
+          cleanup();
+        },
+        0,
+        Math.max(MINIMUM_CLEANUP_PERIOD, SESSION_TIMEOUT / 5),
+        TimeUnit.MILLISECONDS);
+  }
+
+  public void register(long id) {
+    if (SESSION_TIMEOUT == 0) {
+      return;
+    }
+
+    sessionIdToLastActiveTime.put(id, System.currentTimeMillis());
+  }
+
+  public boolean unregister(long id) {
+    if (SESSION_TIMEOUT == 0) {
+      return SessionManager.getInstance().releaseSessionResource(id);
+    }
+
+    if (SessionManager.getInstance().releaseSessionResource(id)) {
+      return sessionIdToLastActiveTime.remove(id) != null;
+    }
+
+    return false;
+  }
+
+  public void refresh(long id) {
+    if (SESSION_TIMEOUT == 0) {
+      return;
+    }
+
+    sessionIdToLastActiveTime.computeIfPresent(id, (k, v) -> System.currentTimeMillis());
+  }
+
+  private void cleanup() {
+    long currentTime = System.currentTimeMillis();
+    sessionIdToLastActiveTime.entrySet().stream()
+        .filter(entry -> entry.getValue() + SESSION_TIMEOUT < currentTime)
+        .forEach(
+            entry -> {
+              if (unregister(entry.getKey())) {
+                LOGGER.debug(
+                    String.format(
+                        "session-%s timed out in %d ms",
+                        entry.getKey(), currentTime - entry.getValue()));
+              }
+            });
+  }
+
+  public static SessionTimeoutManager getInstance() {
+    return SessionTimeoutManagerHelper.INSTANCE;
+  }
+
+  private static class SessionTimeoutManagerHelper {
+
+    private static final SessionTimeoutManager INSTANCE = new SessionTimeoutManager();
+
+    private SessionTimeoutManagerHelper() {}
+  }
+}