refactor(query): remove unsafe implementation of Send and Sync for AcquireQueueGuard (#17818)

The following Send/Sync implementation does not provide any safety
guarantee.

```
unsafe impl Send for AcquireQueueGuard {}
unsafe impl Sync for AcquireQueueGuard {}
```

In this commit, by removing potential `Sync` access to internal data in
`HttpQuery`, `AcquireQueueGuard` does not need to be `Sync` any more.
Therefore there is no need to add unsafe Send/Sync impl to it.

Because `Arc<T>: Send` requires `T: Sync`, but `Arc<Mutex<T>>: Send`
does not require `T: Sync`, we replace `Arc<T>` with `Arc<Mutex<T>>`,
to eliminate the `Sync` requirement.

Therefore, `AcquireQueueGuard` is only required to be `Send`.

Author: drmingdrmer

src/query/service/src/servers/http/v1/http_query_handlers.rs

@@ -273,7 +273,6 @@ async fn query_final_handler(
             Some(query) => {
                 let mut response = query
                     .get_response_state_only()
-                    .await
                     .map_err(HttpErrorCode::server_error)?;
                 // it is safe to set these 2 fields to None, because client now check for null/None first.
                 response.session = None;
@@ -338,7 +337,6 @@ async fn query_state_handler(
                 } else {
                     let response = query
                         .get_response_state_only()
-                        .await
                         .map_err(HttpErrorCode::server_error)?;
                     Ok(QueryResponse::from_internal(query_id, response, false))
                 }

src/query/service/src/servers/http/v1/query/execute_state.rs

@@ -16,7 +16,6 @@ use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::SystemTime;
 
-use databend_common_base::base::tokio::sync::RwLock;
 use databend_common_base::base::ProgressValues;
 use databend_common_base::base::SpillProgress;
 use databend_common_base::runtime::CatchUnwindFuture;
@@ -34,6 +33,7 @@ use futures::StreamExt;
 use log::debug;
 use log::error;
 use log::info;
+use parking_lot::Mutex;
 use serde::Deserialize;
 use serde::Serialize;
 use ExecuteState::*;
@@ -245,25 +245,23 @@ impl Executor {
         }
     }
 
-    #[async_backtrace::framed]
-    pub async fn start_to_running(this: &Arc<RwLock<Executor>>, state: ExecuteState) {
-        let mut guard = this.write().await;
+    pub fn start_to_running(this: &Arc<Mutex<Executor>>, state: ExecuteState) {
+        let mut guard = this.lock();
         if let Starting(_) = &guard.state {
             guard.state = state
         }
     }
 
-    #[async_backtrace::framed]
-    pub async fn start_to_stop(this: &Arc<RwLock<Executor>>, state: ExecuteState) {
-        let mut guard = this.write().await;
+    pub fn start_to_stop(this: &Arc<Mutex<Executor>>, state: ExecuteState) {
+        let mut guard = this.lock();
         if let Starting(_) = &guard.state {
             guard.state = state
         }
     }
-    #[async_backtrace::framed]
-    pub async fn stop<C>(this: &Arc<RwLock<Executor>>, reason: Result<(), C>) {
+
+    pub fn stop<C>(this: &Arc<Mutex<Executor>>, reason: Result<(), C>) {
         let reason = reason.with_context(|| "execution stopped");
-        let mut guard = this.write().await;
+        let mut guard = this.lock();
 
         let state = match &guard.state {
             Starting(s) => {
@@ -337,7 +335,7 @@ impl Executor {
 impl ExecuteState {
     #[async_backtrace::framed]
     pub(crate) async fn try_start_query(
-        executor: Arc<RwLock<Executor>>,
+        executor: Arc<Mutex<Executor>>,
         sql: String,
         session: Arc<Session>,
         ctx: Arc<QueryContext>,
@@ -377,7 +375,7 @@ impl ExecuteState {
             has_result_set,
         };
         info!("http query change state to Running");
-        Executor::start_to_running(&executor, Running(running_state)).await;
+        Executor::start_to_running(&executor, Running(running_state));
 
         let executor_clone = executor.clone();
         let ctx_clone = ctx.clone();
@@ -392,11 +390,11 @@ impl ExecuteState {
         );
         match CatchUnwindFuture::create(res).await {
             Ok(Err(err)) => {
-                Executor::stop(&executor_clone, Err(err.clone())).await;
+                Executor::stop(&executor_clone, Err(err.clone()));
                 block_sender_closer.close();
             }
             Err(e) => {
-                Executor::stop(&executor_clone, Err(e)).await;
+                Executor::stop(&executor_clone, Err(e));
                 block_sender_closer.close();
             }
             _ => {}
@@ -411,7 +409,7 @@ async fn execute(
     schema: DataSchemaRef,
     ctx: Arc<QueryContext>,
     block_sender: SizedChannelSender<DataBlock>,
-    executor: Arc<RwLock<Executor>>,
+    executor: Arc<Mutex<Executor>>,
 ) -> Result<(), ExecutionError> {
     let make_error = || format!("failed to execute {}", interpreter.name());
 
@@ -423,11 +421,11 @@ async fn execute(
         None => {
             let block = DataBlock::empty_with_schema(schema);
             block_sender.send(block, 0).await;
-            Executor::stop::<()>(&executor, Ok(())).await;
+            Executor::stop::<()>(&executor, Ok(()));
             block_sender.close();
         }
         Some(Err(err)) => {
-            Executor::stop(&executor, Err(err)).await;
+            Executor::stop(&executor, Err(err));
             block_sender.close();
         }
         Some(Ok(block)) => {
@@ -444,7 +442,7 @@ async fn execute(
                     }
                 };
             }
-            Executor::stop::<()>(&executor, Ok(())).await;
+            Executor::stop::<()>(&executor, Ok(()));
             block_sender.close();
         }
     }

src/query/service/src/servers/http/v1/query/http_query.rs

@@ -23,7 +23,6 @@ use std::time::Instant;
 
 use databend_common_base::base::short_sql;
 use databend_common_base::base::tokio::sync::Mutex as TokioMutex;
-use databend_common_base::base::tokio::sync::RwLock;
 use databend_common_base::runtime::CatchUnwindFuture;
 use databend_common_base::runtime::GlobalQueryRuntime;
 use databend_common_base::runtime::TrySpawn;
@@ -355,9 +354,9 @@ pub struct HttpQuery {
     pub(crate) session_id: String,
     pub(crate) node_id: String,
     request: HttpQueryRequest,
-    state: Arc<RwLock<Executor>>,
+    state: Arc<Mutex<Executor>>,
     page_manager: Arc<TokioMutex<PageManager>>,
-    expire_state: Arc<parking_lot::Mutex<ExpireState>>,
+    expire_state: Arc<Mutex<ExpireState>>,
     /// The timeout for the query result polling. In the normal case, the client driver
     /// should fetch the paginated result in a timely manner, and the interval should not
     /// exceed this result_timeout_secs.
@@ -514,7 +513,7 @@ impl HttpQuery {
 
         let (block_sender, block_receiver) = sized_spsc(request.pagination.max_rows_in_buffer);
 
-        let state = Arc::new(RwLock::new(Executor {
+        let state = Arc::new(Mutex::new(Executor {
             query_id: query_id.clone(),
             state: ExecuteState::Starting(ExecuteStarting { ctx: ctx.clone() }),
         }));
@@ -565,8 +564,7 @@ impl HttpQuery {
                         warnings: ctx_clone.pop_warnings(),
                     };
                     info!("http query change state to Stopped, fail to start {:?}", e);
-                    Executor::start_to_stop(&state_clone, ExecuteState::Stopped(Box::new(state)))
-                        .await;
+                    Executor::start_to_stop(&state_clone, ExecuteState::Stopped(Box::new(state)));
                     block_sender_closer.close();
                 }
             }
@@ -608,7 +606,7 @@ impl HttpQuery {
     #[fastrace::trace]
     pub async fn get_response_page(&self, page_no: usize) -> Result<HttpQueryResponseInternal> {
         let data = Some(self.get_page(page_no).await?);
-        let state = self.get_state().await;
+        let state = self.get_state();
         let session = self.get_response_session().await?;
 
         Ok(HttpQueryResponseInternal {
@@ -621,9 +619,8 @@ impl HttpQuery {
         })
     }
 
-    #[async_backtrace::framed]
-    pub async fn get_response_state_only(&self) -> Result<HttpQueryResponseInternal> {
-        let state = self.get_state().await;
+    pub fn get_response_state_only(&self) -> Result<HttpQueryResponseInternal> {
+        let state = self.get_state();
 
         Ok(HttpQueryResponseInternal {
             data: None,
@@ -635,9 +632,8 @@ impl HttpQuery {
         })
     }
 
-    #[async_backtrace::framed]
-    async fn get_state(&self) -> ResponseState {
-        let state = self.state.read().await;
+    fn get_state(&self) -> ResponseState {
+        let state = self.state.lock();
         state.get_response_state()
     }
 
@@ -648,8 +644,15 @@ impl HttpQuery {
         // - role: updated by SET ROLE;
         // - secondary_roles: updated by SET SECONDARY ROLES ALL|NONE;
         // - settings: updated by SET XXX = YYY;
-        let executor = self.state.read().await;
-        let session_state = executor.get_session_state();
+
+        let (session_state, is_stopped) = {
+            let executor = self.state.lock();
+
+            let session_state = executor.get_session_state();
+            let is_stopped = matches!(executor.state, ExecuteState::Stopped(_));
+
+            (session_state, is_stopped)
+        };
 
         let settings = session_state
             .settings
@@ -669,7 +672,7 @@ impl HttpQuery {
             None
         };
 
-        if matches!(executor.state, ExecuteState::Stopped(_)) {
+        if is_stopped {
             if let Some(cid) = &self.client_session_id {
                 let (has_temp_table_after_run, just_changed) = {
                     let mut guard = self.has_temp_table_after_run.lock();
@@ -764,7 +767,7 @@ impl HttpQuery {
         // the query will be removed from the query manager before the session is dropped.
         self.detach().await;
 
-        Executor::stop(&self.state, Err(reason)).await;
+        Executor::stop(&self.state, Err(reason));
     }
 
     #[async_backtrace::framed]

src/query/service/src/sessions/queue_mgr.rs

@@ -234,9 +234,6 @@ pub struct AcquireQueueGuard {
     permit: Option<Permit>,
 }
 
-unsafe impl Send for AcquireQueueGuard {}
-unsafe impl Sync for AcquireQueueGuard {}
-
 impl Drop for AcquireQueueGuard {
     fn drop(&mut self) {
         if self.permit.is_some() {