Async Connection Layer #50

IHEII · 2023-05-24T09:07:43Z

What changes were proposed in this pull request?

Client Runtime.
Asynchronous Connections.

Why are the changes needed?

For better performance.

Will break the compatibility? How if so?

Users will currently not be aware of asynchronous changes in the connection layer, and they will use the existing code without changing it.

Does this PR introduce any user-facing change?

Users will currently not be aware of asynchronous changes in the connection layer, and they will use the existing code without changing it.

How was this patch tested?

3.2.4

Checklist

I've run the tests to see all new and existing tests pass.
If this Pull Request resolves an issue, I linked to the issue in the text above.
I've informed the technical writer about the documentation change if necessary.

IHEII · 2023-05-24T09:10:46Z

This MR is the premise of implementing the asynchronous API(#16)

WeiXinChan · 2023-05-24T09:33:25Z

LGTM

WeiXinChan · 2023-05-24T09:35:10Z

src/rpc/mod.rs

@@ -544,42 +540,45 @@ impl Connection {
        let rx = self.send(req, channel_id)?;

        if payload.timeout_millis() == 0 {
-            // no-wait request,return Ok directly
+            // no-wait request,return Ok directly2


directly2 -> directly

ShiKaiWi · 2023-05-24T09:25:47Z

src/rpc/mod.rs

            if let Err(e) = sender.send(Err(CommonErr(
                CommonErrCode::Rpc,
                "connection reader exits".to_owned(),
            ))) {
                error!(
                    "Connection::cancel_requests: fail to send cancel message, err:{}",
-                    e
+                    e.err().unwrap().to_string()


Is this unwrap safe? Can we just print {e:?}?

ShiKaiWi · 2023-05-24T09:29:31Z

src/rpc/mod.rs

@@ -544,42 +540,45 @@ impl Connection {
        let rx = self.send(req, channel_id)?;

        if payload.timeout_millis() == 0 {
-            // no-wait request,return Ok directly
+            // no-wait request,return Ok directly2


Suggested change

// no-wait request,return Ok directly2

// no-wait request,return Ok directly

ShiKaiWi · 2023-05-24T09:30:51Z

src/rpc/mod.rs

-                    error!(
+        // TODO: remove block_on with rx.await
+        let resp = self.runtimes.default_runtime.block_on(async move {
+            match TokioTimeout(timeout, rx).await {


When use a free function from a module, just use <module_name>::<function_name>:

Suggested change

match TokioTimeout(timeout, rx).await {

match tokio::time::timeout(timeout, rx).await {

ShiKaiWi · 2023-05-24T09:32:53Z

src/rpc/mod.rs

+                        format!("wait for rpc response timeout, err:{err}"),
+                    ));
+                }
+            }.expect("Tokio timeout panics, may be there is no current timer set")


For a library, throw up the error instead of panic.

ShiKaiWi · 2023-05-24T09:33:52Z

src/rpc/mod.rs

        //2. close reader
-        if let Err(e) = self
-            .reader_signal_sender
-            .send(())
-            .map_err(ConnectionSender::broken_pipe)
-        {
+        //TODO: remove block_on


// 2. close reader // TODO: remove block_on

ShiKaiWi · 2023-05-24T09:44:43Z

Cargo.toml

@@ -54,6 +54,7 @@ serde = "1.0"
 serde_bytes = "0.11"
 serde_derive = "1.0"
 serde_json = "1.0"
+socket2 = "0.5"


I find that what the tokio::net module uses is just socket2, so I guess there is no need to import it explicitly?

To set some parameters like "tcp_keepalive", and "buffer_size"..., it turns out that we have to use socket2 to define a socket that owns those properties to create a Tokio TCP stream. Socket2 is required explicitly, in order to customize a TCP stream.
Learn more from rust-lang/rust#69774 and rust-lang

ShiKaiWi · 2023-05-24T09:48:06Z

src/client/mod.rs

+            conn_reader_threads: 6,
+            conn_writer_threads: 4,
+            default_threads_num: 2,


Why not keep them in the same style? And how about setting all these runtime's thread number is 2 by default?

Suggested change

conn_reader_threads: 6,

conn_writer_threads: 4,

default_threads_num: 2,

conn_reader_thread_num: 2,

conn_writer_thread_num: 2,

default_thread_num: 2,

Perhaps I could modify it into 422. The reason why I set reader threads into 4 is that there are more "read/get tasks" in practice. After I test this client on the YCSB, the number of reader threads and the number of connections are the bottleneck of the client. So 422 is a better choice, which implies that the count of reader threads should be more than others.

Rachelint · 2023-05-24T10:06:21Z

src/client/table_client.rs

 /// OBKV Table client
 #[derive(Clone)]
+#[allow(dead_code)]


Why dead code?

Dead_code is for runtimes. Runtimes that we add in this commit are for async refresh_thread_pool, which have not been implemented yet.

Rachelint · 2023-05-24T10:07:04Z

src/client/table_client.rs

@@ -193,6 +196,9 @@ struct ObTableClientInner {
    closed: AtomicBool,
    status_mutex: Lock,

+    /// Client Runtimes
+    runtimes: RuntimesRef,
+
    //ServerAddr(all) -> ObTableConnection


"//" -> "// "

Rachelint · 2023-05-24T11:33:40Z

src/rpc/mod.rs

    ) -> ConnectionSender {
-        let (sender, receiver): (Sender<ObTablePacket>, Receiver<ObTablePacket>) = unbounded();
+        let (sender, mut receiver): (mpsc::Sender<ObTablePacket>, mpsc::Receiver<ObTablePacket>) =
+            mpsc::channel(100);


Make it configurable?

Rachelint · 2023-05-24T11:44:58Z

src/rpc/mod.rs

                            break;
                        }
                    }
                }

                active.store(false, Ordering::Release);

-                if let Err(err) = write_stream.shutdown(Shutdown::Write) {
+                if let Err(err) = write_stream.shutdown().await {
                    error!("Fail to close write stream to {}, err:{}", addr, err);


As I know, such as error!("Fail to close write stream to {addr}, err:{err}") is supported now?

Rachelint · 2023-05-25T05:46:33Z

src/rpc/mod.rs

@@ -302,12 +307,13 @@ impl Connection {
        let mut buf = BytesMut::with_capacity(READ_BUF_SIZE);
        loop {
            if let Ok(()) = signal_receiver.try_recv() {
+                error!("DEBUG: out0");


Make it to debug! and refactor the logged message.

Rachelint · 2023-05-25T05:48:28Z

src/rpc/mod.rs

-            );
-        }
+        // TODO: stream close
+        drop(read_stream);


Why not shutdown like write_stream?

From document of OwnedWriteHalf, shutdown is the interface of AsyncWrite, it also mentions that "Dropping the write half will also shut down the write half of the TCP stream. ".
read_stream is an OwnedReadHalf, which means it does not have the shutdown interface. Learn more from OwnedReadHalf
By the way, it seems that tokio users never care about GC on stream. It is still a problem, so I leave a TODO here.

No need to drop I think, when the function end, it will be dropped automatically.
In fact this is the classic RAII.

Make sense, done.

IHEII

Fix review

IHEII · 2023-05-25T02:27:45Z

Cargo.toml

@@ -54,6 +54,7 @@ serde = "1.0"
 serde_bytes = "0.11"
 serde_derive = "1.0"
 serde_json = "1.0"
+socket2 = "0.5"


To set some parameters like "tcp_keepalive", and "buffer_size"..., it turns out that we have to use socket2 to define a socket that owns those properties to create a Tokio TCP stream. Socket2 is required explicitly, in order to customize a TCP stream.
Learn more from rust-lang/rust#69774 and rust-lang

IHEII · 2023-05-25T02:39:55Z

src/client/mod.rs

+            conn_reader_threads: 6,
+            conn_writer_threads: 4,
+            default_threads_num: 2,


Perhaps I could modify it into 422. The reason why I set reader threads into 4 is that there are more "read/get tasks" in practice. After I test this client on the YCSB, the number of reader threads and the number of connections are the bottleneck of the client. So 422 is a better choice, which implies that the count of reader threads should be more than others.

IHEII · 2023-05-25T02:45:19Z

src/client/table_client.rs

@@ -193,6 +196,9 @@ struct ObTableClientInner {
    closed: AtomicBool,
    status_mutex: Lock,

+    /// Client Runtimes
+    runtimes: RuntimesRef,
+
    //ServerAddr(all) -> ObTableConnection


IHEII · 2023-05-25T02:50:35Z

src/client/table_client.rs

 /// OBKV Table client
 #[derive(Clone)]
+#[allow(dead_code)]


Dead_code is for runtimes. Runtimes that we add in this commit are for async refresh_thread_pool, which have not been implemented yet.

IHEII · 2023-05-25T02:56:06Z

src/rpc/mod.rs

            if let Err(e) = sender.send(Err(CommonErr(
                CommonErrCode::Rpc,
                "connection reader exits".to_owned(),
            ))) {
                error!(
                    "Connection::cancel_requests: fail to send cancel message, err:{}",
-                    e
+                    e.err().unwrap().to_string()


IHEII · 2023-05-25T03:15:01Z

src/rpc/mod.rs

        //2. close reader
-        if let Err(e) = self
-            .reader_signal_sender
-            .send(())
-            .map_err(ConnectionSender::broken_pipe)
-        {
+        //TODO: remove block_on


IHEII · 2023-05-25T05:57:53Z

src/rpc/mod.rs

@@ -302,12 +307,13 @@ impl Connection {
        let mut buf = BytesMut::with_capacity(READ_BUF_SIZE);
        loop {
            if let Ok(()) = signal_receiver.try_recv() {
+                error!("DEBUG: out0");


IHEII · 2023-05-25T06:58:06Z

src/rpc/mod.rs

-            );
-        }
+        // TODO: stream close
+        drop(read_stream);


From document of OwnedWriteHalf, shutdown is the interface of AsyncWrite, it also mentions that "Dropping the write half will also shut down the write half of the TCP stream. ".
read_stream is an OwnedReadHalf, which means it does not have the shutdown interface. Learn more from OwnedReadHalf
By the way, it seems that tokio users never care about GC on stream. It is still a problem, so I leave a TODO here.

IHEII · 2023-05-25T07:06:10Z

src/rpc/mod.rs

    ) -> ConnectionSender {
-        let (sender, receiver): (Sender<ObTablePacket>, Receiver<ObTablePacket>) = unbounded();
+        let (sender, mut receiver): (mpsc::Sender<ObTablePacket>, mpsc::Receiver<ObTablePacket>) =
+            mpsc::channel(100);


IHEII · 2023-05-25T07:06:54Z

src/rpc/mod.rs

                            break;
                        }
                    }
                }

                active.store(false, Ordering::Release);

-                if let Err(err) = write_stream.shutdown(Shutdown::Write) {
+                if let Err(err) = write_stream.shutdown().await {
                    error!("Fail to close write stream to {}, err:{}", addr, err);


IHEII

Fix review

ShiKaiWi · 2023-05-25T08:05:14Z

src/client/mod.rs

+    pub conn_reader_threads: usize,
+    pub conn_writer_threads: usize,
+    pub default_threads_num: usize,


Shall we keep the names of these three config options in the same style:

Suggested change

pub conn_reader_threads: usize,

pub conn_writer_threads: usize,

pub default_threads_num: usize,

pub conn_reader_thread_num: usize,

pub conn_writer_thread_num: usize,

pub default_thread_num: usize,

Rachelint · 2023-05-29T02:52:35Z

tests/test_table_client_base.rs

@@ -34,7 +34,7 @@ pub struct BaseTest {
 }

 impl BaseTest {
-    const ROW_NUM: usize = 500;
+    const ROW_NUM: usize = 400;


Why modify to 400?

When the client and the observer are not in the same place, the latency will make this test spend too much time. Reduce the count of rows to reduce the runtime.

Rachelint · 2023-05-29T08:23:37Z

src/rpc/mod.rs

-            );
-        }
+        // TODO: stream close
+        drop(read_stream);


No need to drop I think, when the function end, it will be dropped automatically.
In fact this is the classic RAII.

IHEII

Fix review

IHEII · 2023-05-29T08:48:36Z

tests/test_table_client_base.rs

@@ -34,7 +34,7 @@ pub struct BaseTest {
 }

 impl BaseTest {
-    const ROW_NUM: usize = 500;
+    const ROW_NUM: usize = 400;


When the client and the observer are not in the same place, the latency will make this test spend too much time. Reduce the count of rows to reduce the runtime.

IHEII · 2023-05-29T08:50:27Z

src/rpc/mod.rs

-            );
-        }
+        // TODO: stream close
+        drop(read_stream);


Make sense, done.

IHEII · 2023-05-29T08:50:48Z

src/client/mod.rs

+    pub conn_reader_threads: usize,
+    pub conn_writer_threads: usize,
+    pub default_threads_num: usize,


ShiKaiWi · 2023-05-29T11:05:11Z

src/client/mod.rs

+    pub conn_writer_thread_num: usize,
+    pub default_thread_num: usize,
+
+    pub conn_sender_mpsc_channel: usize,


Maybe we need a more readable name for this option:

Suggested change

pub conn_sender_mpsc_channel: usize,

pub max_reqs_in_flight_per_conn: usize,

Done, using max_inflight_reqs_per_conn instead.

ShiKaiWi

LGTM

IHEII added 2 commits May 19, 2023 18:33

[Feat] client runtime

f41a236

[Feat] asynchronous connections

78e5b54

IHEII linked an issue May 24, 2023 that may be closed by this pull request

Support async API #16

Closed

2 tasks

IHEII removed a link to an issue May 24, 2023

Support async API #16

Closed

2 tasks

IHEII changed the title ~~Async Connection~~ Async Connection Layer May 24, 2023

WeiXinChan reviewed May 24, 2023

View reviewed changes

ShiKaiWi reviewed May 24, 2023

View reviewed changes

Rachelint approved these changes May 24, 2023

View reviewed changes

Rachelint reviewed May 25, 2023

View reviewed changes

[Fix] review

0fffa9f

IHEII commented May 25, 2023

View reviewed changes

ShiKaiWi reviewed May 25, 2023

View reviewed changes

Rachelint reviewed May 29, 2023

View reviewed changes

[Fix] review

8b7dedf

IHEII commented May 29, 2023

View reviewed changes

ShiKaiWi reviewed May 29, 2023

View reviewed changes

[Fix] review

00583e5

ShiKaiWi approved these changes May 29, 2023

View reviewed changes

IHEII merged commit 7bcf0c5 into oceanbase:AsyncDev May 29, 2023

IHEII deleted the async branch May 30, 2023 01:59

	// no-wait request,return Ok directly2
	// no-wait request,return Ok directly

	match TokioTimeout(timeout, rx).await {
	match tokio::time::timeout(timeout, rx).await {

	pub conn_sender_mpsc_channel: usize,
	pub max_reqs_in_flight_per_conn: usize,

Async Connection Layer #50

Async Connection Layer #50

Conversation

IHEII commented May 24, 2023 • edited Loading

What changes were proposed in this pull request?

Why are the changes needed?

Will break the compatibility? How if so?

Does this PR introduce any user-facing change?

How was this patch tested?

Checklist

IHEII commented May 24, 2023

WeiXinChan commented May 24, 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

ShiKaiWi May 24, 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

ShiKaiWi May 24, 2023 • edited Loading

Choose a reason for hiding this comment

IHEII May 25, 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

IHEII left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

IHEII May 25, 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

IHEII left a 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 May 29, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

IHEII left a comment

Choose a reason for hiding this comment

IHEII May 29, 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

ShiKaiWi left a comment

Choose a reason for hiding this comment

IHEII commented May 24, 2023 •

edited

Loading

ShiKaiWi May 24, 2023 •

edited

Loading

ShiKaiWi May 24, 2023 •

edited

Loading

IHEII May 25, 2023 •

edited

Loading

IHEII May 25, 2023 •

edited

Loading

IHEII May 29, 2023 •

edited

Loading

IHEII May 29, 2023 •

edited

Loading