refactor(sessions): remove useless code async

Author: zhang2014

src/query/catalog/src/table_context.rs

@@ -103,5 +103,5 @@ pub trait TableContext: Send + Sync {
     fn get_cluster(&self) -> Arc<Cluster>;
     async fn get_table(&self, catalog: &str, database: &str, table: &str)
     -> Result<Arc<dyn Table>>;
-    async fn get_processes_info(&self) -> Vec<ProcessInfo>;
+    fn get_processes_info(&self) -> Vec<ProcessInfo>;
 }

src/query/service/src/interpreters/interpreter_kill.rs

@@ -35,7 +35,7 @@ impl KillInterpreter {
     }
 
     async fn execute_kill(&self, session_id: &String) -> Result<PipelineBuildResult> {
-        match self.ctx.get_session_by_id(session_id).await {
+        match self.ctx.get_session_by_id(session_id) {
             None => Err(ErrorCode::UnknownSession(format!(
                 "Not found session id {}",
                 session_id
@@ -73,16 +73,13 @@ impl Interpreter for KillInterpreter {
         // otherwise use the session_id.
         // More info Link to: https://github.com/datafuselabs/databend/discussions/5405.
         match id.parse::<u32>() {
-            Ok(mysql_conn_id) => {
-                let session_id = self.ctx.get_id_by_mysql_conn_id(&Some(mysql_conn_id)).await;
-                match session_id {
-                    Some(get) => self.execute_kill(&get).await,
-                    None => Err(ErrorCode::UnknownSession(format!(
-                        "MySQL connection id {} not found session id",
-                        mysql_conn_id
-                    ))),
-                }
-            }
+            Ok(mysql_conn_id) => match self.ctx.get_id_by_mysql_conn_id(&Some(mysql_conn_id)) {
+                Some(get) => self.execute_kill(&get).await,
+                None => Err(ErrorCode::UnknownSession(format!(
+                    "MySQL connection id {} not found session id",
+                    mysql_conn_id
+                ))),
+            },
             Err(_) => self.execute_kill(id).await,
         }
     }

src/query/service/src/sessions/query_ctx.rs

@@ -171,23 +171,18 @@ impl QueryContext {
     }
 
     // Get one session by session id.
-    pub async fn get_session_by_id(self: &Arc<Self>, id: &str) -> Option<Arc<Session>> {
-        SessionManager::instance().get_session_by_id(id).await
+    pub fn get_session_by_id(self: &Arc<Self>, id: &str) -> Option<Arc<Session>> {
+        SessionManager::instance().get_session_by_id(id)
     }
 
     // Get session id by mysql connection id.
-    pub async fn get_id_by_mysql_conn_id(
-        self: &Arc<Self>,
-        conn_id: &Option<u32>,
-    ) -> Option<String> {
-        SessionManager::instance()
-            .get_id_by_mysql_conn_id(conn_id)
-            .await
+    pub fn get_id_by_mysql_conn_id(self: &Arc<Self>, conn_id: &Option<u32>) -> Option<String> {
+        SessionManager::instance().get_id_by_mysql_conn_id(conn_id)
     }
 
     // Get all the processes list info.
-    pub async fn get_processes_info(self: &Arc<Self>) -> Vec<ProcessInfo> {
-        SessionManager::instance().processes_info().await
+    pub fn get_processes_info(self: &Arc<Self>) -> Vec<ProcessInfo> {
+        SessionManager::instance().processes_info()
     }
 
     /// Get the client socket address.
@@ -392,8 +387,8 @@ impl TableContext for QueryContext {
     }
 
     // Get all the processes list info.
-    async fn get_processes_info(&self) -> Vec<ProcessInfo> {
-        SessionManager::instance().processes_info().await
+    fn get_processes_info(&self) -> Vec<ProcessInfo> {
+        SessionManager::instance().processes_info()
     }
 }
 

src/query/service/src/sessions/session_mgr.rs

@@ -147,12 +147,12 @@ impl SessionManager {
         }
     }
 
-    pub async fn get_session_by_id(&self, id: &str) -> Option<Arc<Session>> {
+    pub fn get_session_by_id(&self, id: &str) -> Option<Arc<Session>> {
         let sessions = self.active_sessions.read();
         sessions.get(id).and_then(|weak_ptr| weak_ptr.upgrade())
     }
 
-    pub async fn get_id_by_mysql_conn_id(&self, mysql_conn_id: &Option<u32>) -> Option<String> {
+    pub fn get_id_by_mysql_conn_id(&self, mysql_conn_id: &Option<u32>) -> Option<String> {
         let sessions = self.mysql_conn_map.read();
         sessions.get(mysql_conn_id).cloned()
     }
@@ -194,7 +194,7 @@ impl SessionManager {
             let mut signal = Box::pin(signal.next());
 
             for _index in 0..timeout_secs {
-                if SessionManager::destroy_idle_sessions(&active_sessions).await {
+                if SessionManager::destroy_idle_sessions(&active_sessions) {
                     return;
                 }
 
@@ -216,7 +216,7 @@ impl SessionManager {
         }
     }
 
-    pub async fn processes_info(&self) -> Vec<ProcessInfo> {
+    pub fn processes_info(&self) -> Vec<ProcessInfo> {
         let sessions = self.active_sessions.read();
 
         let mut processes_info = Vec::with_capacity(sessions.len());
@@ -229,7 +229,7 @@ impl SessionManager {
         processes_info
     }
 
-    async fn destroy_idle_sessions(sessions: &Arc<RwLock<HashMap<String, Weak<Session>>>>) -> bool {
+    fn destroy_idle_sessions(sessions: &Arc<RwLock<HashMap<String, Weak<Session>>>>) -> bool {
         // Read lock does not support reentrant
         // https://github.com/Amanieu/parking_lot::/blob/lock_api-0.4.4/lock_api/src/rwlock.rs#L422
         let mut active_sessions_read_guard = sessions.write();

src/query/storages/preludes/src/system/processes_table.rs

@@ -26,24 +26,24 @@ use common_meta_app::schema::TableMeta;
 use common_meta_types::UserInfo;
 
 use crate::sessions::TableContext;
-use crate::storages::system::table::AsyncOneBlockSystemTable;
-use crate::storages::system::table::AsyncSystemTable;
 use crate::storages::Table;
+use crate::system::SyncOneBlockSystemTable;
+use crate::system::SyncSystemTable;
 
 pub struct ProcessesTable {
     table_info: TableInfo,
 }
 
 #[async_trait::async_trait]
-impl AsyncSystemTable for ProcessesTable {
+impl SyncSystemTable for ProcessesTable {
     const NAME: &'static str = "system.processes";
 
     fn get_table_info(&self) -> &TableInfo {
         &self.table_info
     }
 
-    async fn get_full_data(&self, ctx: Arc<dyn TableContext>) -> Result<DataBlock> {
-        let processes_info = ctx.get_processes_info().await;
+    fn get_full_data(&self, ctx: Arc<dyn TableContext>) -> Result<DataBlock> {
+        let processes_info = ctx.get_processes_info();
 
         let mut processes_id = Vec::with_capacity(processes_info.len());
         let mut processes_type = Vec::with_capacity(processes_info.len());
@@ -129,7 +129,7 @@ impl ProcessesTable {
             },
         };
 
-        AsyncOneBlockSystemTable::create(ProcessesTable { table_info })
+        SyncOneBlockSystemTable::create(ProcessesTable { table_info })
     }
 
     fn process_host(client_address: &Option<SocketAddr>) -> Option<Vec<u8>> {