[Feat] Async Interface #52

IHEII · 2023-06-06T08:06:16Z

Task Description

Implement the asynchronous interface for the OBKV client.

Solution Description

Fix Support async API #16, support asynchronous API by using Tokio.

Passed Regressions

Pass all tests in 3.x

Upgrade Compatibility

Synchronous API may not be provided, but users could find it on branch SyncInterface, and we are considering adding those Synchronous APIs into the client.

Other Information

Release Note

Impl async APIs

ShiKaiWi · 2023-06-06T11:11:33Z

src/client/query.rs

+        if self.closed {
+        } else {
+            error!("QueryStreamResult::close fail")
        }


Suggested change

if self.closed {

} else {

error!("QueryStreamResult::close fail")

}

if !self.closed {

error!("QueryStreamResult::drop stream is not closed when drop")

}

ShiKaiWi · 2023-06-06T11:13:04Z

src/client/query.rs


-impl Iterator for QueryResultSet {
-    type Item = Result<HashMap<String, Value>>;
+    pub async fn async_close(&mut self) -> Result<()> {


Why not remove the close method, and name this method as close?

Fixed, only keep the async close now.

ShiKaiWi · 2023-06-06T11:14:53Z

src/client/table.rs

    }

    pub fn operation_timeout(&self) -> Duration {
        self.config.rpc_operation_timeout
    }

-    fn execute(
+    /// Execute a batch operation on a table
+    pub async fn execute_batch(


I guess we should move this method to the obclient if all the other methods are removed from the ObTable.

Only keep the methods used by ObTableClient now, execute_batch() is used in batch operation, and execute() is used in regular operation.
When we use get(), we actually get the data from an observer. ObTable is an abstraction of the observer, so every request needs to be executed by the ObTable.

ShiKaiWi · 2023-06-06T11:21:42Z

src/client/table.rs

-}
-
+#[allow(dead_code)]
+// impl ObTableStreamQuerier for obtable


Can these traits including ObTableStreamQuerier, Table and StreamQuerier be removed? I remember that only just one implementation for them, and they are too complex to be mocked if we want to use them in unit tests.

Fixed, only keep the ObTableClientStreamQuerier now, and rename it into StreamQuerier in case to keep the code clean.

ShiKaiWi · 2023-06-06T11:39:09Z

src/rpc/conn_pool.rs

@@ -249,7 +246,7 @@ impl ConnPool {
        Ok(())
    }

-    pub fn get(&self) -> Result<Arc<Connection>> {
+    pub async fn get(&self) -> Result<Arc<Connection>> {


Is this async necessary? No await is found in the code block of get.

ShiKaiWi · 2023-06-06T11:51:24Z

src/rpc/conn_pool.rs

@@ -319,42 +316,42 @@ struct SharedPool {
    conn_builder: ConnBuilder,
    inner: Mutex<PoolInner>,
    cond: Condvar,
-    conn_init_thread_pool: Arc<ScheduledThreadPool>,
+    runtimes: RuntimesRef,


Just provide a specific runtime is enough? I guess we can call it as bg_runtime?

Fixed. We reduce the runtime (learn more from ObClientRuntimes), and use bg_runtime here.

ShiKaiWi · 2023-06-06T11:52:37Z

src/rpc/mod.rs

@@ -679,24 +672,29 @@ impl Connection {
    /// invalidated.
    ///
    ///For info on default settings see [Builder](struct.Builder.html)
-    pub fn new() -> Result<Connection> {
-        Builder::new().build()
+    pub async fn new() -> Result<Connection> {


try_new is better.

ShiKaiWi · 2023-06-06T11:55:36Z

src/client/table_client.rs

+        batch_op_runtime: Arc::new(build_runtime(
+            "ob-batch-executor",
+            config.batch_op_thread_num,
+        )),


I guess this runtime is not necessary in the async context, how about remove it?

Fixed, we use bg_runtime to deal with a batch operation. It is necessary to create a runtime to process those requests, since batch operation may operate data from different partitions and servers. Hence, we need to seed packages simultaneously and aggregate the results.

ShiKaiWi · 2023-06-06T11:55:54Z

src/client/table_client.rs

+            "ob-batch-executor",
+            config.batch_op_thread_num,
+        )),
+        query_runtime: Arc::new(build_runtime("ob-query-executor", config.query_thread_num)),


Maybe we should use reader_runtime?

ShiKaiWi · 2023-06-06T11:57:32Z

src/client/table_client.rs

+            config.batch_op_thread_num,
+        )),
+        query_runtime: Arc::new(build_runtime("ob-query-executor", config.query_thread_num)),
+        conn_init_runtime: Arc::new(build_runtime("ob-conn-initer", config.conn_init_thread_num)),


Maybe what we need here is just a bg_runtime to do some work including connection initialization, table location refreshing?

And replace default_runtime with the bg_runtime.

Fixed. We only keep three necessary runtimes for different tasks. See ObClientRuntimes

ShiKaiWi · 2023-06-12T03:18:02Z

src/client/table_client.rs

-        default_runtime: Arc::new(build_runtime("ob-default", config.default_thread_num)),
+        tcp_recv_runtime: Arc::new(build_runtime("ob-tcp-reviever", config.tcp_recv_thread_num)),
+        tcp_send_runtime: Arc::new(build_runtime("ob-tcp-sender", config.tcp_send_thread_num)),
+        bg_runtime: Arc::new(build_runtime("ob_bg", config.bg_thread_num)),


Suggested change

bg_runtime: Arc::new(build_runtime("ob_bg", config.bg_thread_num)),

bg_runtime: Arc::new(build_runtime("ob-bg", config.bg_thread_num)),

ShiKaiWi · 2023-06-12T03:21:20Z

LGTM

ShiKaiWi · 2023-06-13T02:49:52Z

src/rpc/mod.rs

+    }
+
+    /// Shutdown the sender without closing remote
+    fn shutdown(&mut self) -> Result<()> {
        let writer = mem::replace(&mut self.writer, None);
        let drop_helper = AbortOnDropMany(vec![writer.unwrap()]);
        drop(drop_helper);


It seems no need to drop manually.

ShiKaiWi · 2023-06-13T02:54:05Z

src/rpc/mod.rs

+    fn shutdown(&mut self) -> Result<()> {
        let writer = mem::replace(&mut self.writer, None);
        let drop_helper = AbortOnDropMany(vec![writer.unwrap()]);


I guess it will be better if you put it as:

if let Some(writer) = mem::take(&mut self.writer) { writer.abort(); }

ShiKaiWi · 2023-06-13T03:40:56Z

src/rpc/mod.rs

+            }
+        }
+
+        let reader = mem::replace(&mut self.reader, None);


mem::take is better.

ShiKaiWi · 2023-06-13T03:52:43Z

src/rpc/mod.rs

    async fn close(&mut self) -> Result<()> {
        self.request(ObTablePacket::ClosePoison).await?;
+        self.shutdown()


If we want to close the ConnectionSender gracefully, I guess we should join the writer handle after sending the poison signal.

Fixed, the writer will handle ObTablePacket::ClosePoison packet, and will no longer receive packets from mpsc::channel. Then JoinHandle writer will drop automatically, all we need to do is wait.

ShiKaiWi · 2023-06-13T11:26:12Z

src/rpc/mod.rs

-        let drop_helper = AbortOnDropMany(vec![writer.unwrap()]);
-        drop(drop_helper);
+        if let Some(writer) = mem::take(&mut self.writer) {
+            let _drop_helper = AbortOnDropMany(vec![writer]);


writer.abort() is enough to avoid one more memory allocation for AbortOnDropMany.

ShiKaiWi · 2023-06-13T11:28:19Z

src/rpc/mod.rs

+        if let Some(writer) = mem::take(&mut self.writer) {
+            writer.await??
+        }
+        self.sender.closed().await;


I guess there is no need to check wait for sender to be closed? What we cares about is whether the tcp send worker is stopped.

IHEII added 2 commits June 5, 2023 17:24

[Feat] async client

96a0cf5

[Feat] async client

8f1304b

IHEII requested review from WeiXinChan and shenyunlong June 6, 2023 08:08

ShiKaiWi reviewed Jun 6, 2023

View reviewed changes

IHEII added 4 commits June 7, 2023 20:51

[Fix] remove useless APIs

e9c1c08

[Fix] modify runtime/query

5164c23

[Feat] batch ycsb

7805abe

[Fix] review

38d6950

ShiKaiWi reviewed Jun 12, 2023

View reviewed changes

[Fix] connection close process

54a62fc

ShiKaiWi reviewed Jun 13, 2023

View reviewed changes

[Fix] remove trait Table & fix review

f136b21

ShiKaiWi reviewed Jun 13, 2023

View reviewed changes

IHEII force-pushed the async branch from 62146cd to f136b21 Compare June 13, 2023 13:27

[Fix] review

ed7c4bf

IHEII merged commit e13051d into oceanbase:AsyncDev Jun 13, 2023

IHEII deleted the async branch June 14, 2023 01:34

IHEII restored the async branch June 14, 2023 01:34

IHEII deleted the async branch June 14, 2023 06:41

	bg_runtime: Arc::new(build_runtime("ob_bg", config.bg_thread_num)),
	bg_runtime: Arc::new(build_runtime("ob-bg", config.bg_thread_num)),

[Feat] Async Interface #52

[Feat] Async Interface #52

Conversation

IHEII commented Jun 6, 2023

Task Description

Solution Description

Passed Regressions

Upgrade Compatibility

Other Information

Release Note

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

IHEII Jun 9, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ShiKaiWi commented Jun 12, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

IHEII Jun 9, 2023 •

edited

Loading