fix: check send error

Author: caiweiwei.cww

.gitignore

@@ -2,4 +2,4 @@ target/
 **/*.rs.bk
 Cargo.lock
 .vscode/
-
+/flamegraph.svg

examples/Cargo.toml

@@ -10,6 +10,7 @@ log = "0.4.13"
 env_logger = "0.8.2"
 futures = "0.3.10"
 clap = "2.33.3"
+pprof = { version = "0.4", features = ["flamegraph"] }
 
 [dev-dependencies.rsocket_rust]
 path = "../rsocket"
@@ -22,7 +23,7 @@ features = ["tls"]
 path = "../rsocket-transport-websocket"
 
 [dev-dependencies.tokio]
-version = "1.0.1"
+version = "1.0.3"
 default-features = false
 features = ["full"]
 

examples/qps.rs

@@ -1,6 +1,7 @@
 #[macro_use]
 extern crate log;
 
+use std::fs::File;
 use std::sync::{
     atomic::{AtomicU32, Ordering},
     Arc,
@@ -65,6 +66,13 @@ async fn main() -> Result<()> {
         .version("0.0.0")
         .author("Jeffsky <jjeffcaii@outlook.com>")
         .about("An QPS benchmark tool for RSocket.")
+        .arg(
+            Arg::with_name("pprof")
+                .long("pprof")
+                .required(false)
+                .takes_value(false)
+                .help("Enable pprof."),
+        )
         .arg(
             Arg::with_name("count")
                 .short("c")
@@ -100,6 +108,12 @@ async fn main() -> Result<()> {
         .value_of("size")
         .map(|s| s.parse().expect("Invalid size!"))
         .unwrap();
+
+    let mut guard = None;
+    if cli.is_present("pprof") {
+        guard = Some(pprof::ProfilerGuard::new(100).unwrap());
+    }
+
     let addr = cli.value_of("URL").unwrap();
 
     let notify = Arc::new(Notify::new());
@@ -145,5 +159,16 @@ async fn main() -> Result<()> {
         costs,
         1000f64 * (count as f64) / (costs as f64)
     );
+
+    if let Some(guard) = guard {
+        match guard.report().build() {
+            Ok(report) => {
+                let file = File::create("flamegraph.svg").unwrap();
+                report.flamegraph(file).unwrap();
+            }
+            Err(_) => {}
+        };
+    }
+
     Ok(())
 }

justfile

@@ -13,4 +13,4 @@ fmt:
 echo:
     @RUST_LOG=release cargo run --release --example echo -- serve tcp://127.0.0.1:7878
 bench:
-    @RUST_LOG=info cargo run --release --example qps -- -c 1000000 -s 1024 tcp://127.0.0.1:7878
+    @RUST_LOG=info cargo run --release --example qps -- -c 1000000 -s 1024 --pprof tcp://127.0.0.1:7878

rsocket-messaging/src/requester.rs

@@ -155,9 +155,9 @@ impl RequesterBuilder {
             Some(TransportKind::TCP(h, p)) => {
                 let addr: SocketAddr = format!("{}:{}", h, p).parse()?;
                 let cli = RSocketFactory::connect()
-                    .data_mime_type(data_mime_type.as_ref())
+                    .data_mime_type(data_mime_type)
                     .setup(setup)
-                    .metadata_mime_type(MimeType::MESSAGE_X_RSOCKET_COMPOSITE_METADATA_V0.as_ref())
+                    .metadata_mime_type(MimeType::MESSAGE_X_RSOCKET_COMPOSITE_METADATA_V0)
                     .transport(TcpClientTransport::from(addr))
                     .start()
                     .await?;
@@ -168,9 +168,9 @@ impl RequesterBuilder {
             Some(TransportKind::WS(u)) => {
                 let url = Url::parse(&u)?;
                 let cli = RSocketFactory::connect()
-                    .data_mime_type(data_mime_type.as_ref())
+                    .data_mime_type(data_mime_type)
                     .setup(setup)
-                    .metadata_mime_type(MimeType::MESSAGE_X_RSOCKET_COMPOSITE_METADATA_V0.as_ref())
+                    .metadata_mime_type(MimeType::MESSAGE_X_RSOCKET_COMPOSITE_METADATA_V0)
                     .transport(WebsocketClientTransport::from(url))
                     .start()
                     .await?;

rsocket/src/core/client.rs

@@ -52,9 +52,10 @@ where
 
     pub fn fragment(mut self, mtu: usize) -> Self {
         if mtu > 0 && mtu < transport::MIN_MTU {
-            panic!("invalid fragment mtu: at least {}!", transport::MIN_MTU)
+            warn!("invalid fragment mtu: at least {}!", transport::MIN_MTU)
+        } else {
+            self.mtu = mtu;
         }
-        self.mtu = mtu;
         self
     }
 
@@ -82,18 +83,22 @@ where
         self
     }
 
-    pub fn mime_type(mut self, metadata_mime_type: &str, data_mime_type: &str) -> Self {
+    pub fn mime_type(
+        mut self,
+        metadata_mime_type: impl Into<String>,
+        data_mime_type: impl Into<String>,
+    ) -> Self {
         self = self.metadata_mime_type(metadata_mime_type);
         self = self.data_mime_type(data_mime_type);
         self
     }
 
-    pub fn data_mime_type(mut self, mime_type: &str) -> Self {
+    pub fn data_mime_type(mut self, mime_type: impl Into<String>) -> Self {
         self.setup = self.setup.set_data_mime_type(mime_type);
         self
     }
 
-    pub fn metadata_mime_type(mut self, mime_type: &str) -> Self {
+    pub fn metadata_mime_type(mut self, mime_type: impl Into<String>) -> Self {
         self.setup = self.setup.set_metadata_mime_type(mime_type);
         self
     }
@@ -232,7 +237,7 @@ where
             }
         });
 
-        socket.setup(setup).await;
+        socket.setup(setup).await?;
 
         Ok(Client::new(socket, close_notify, closing))
     }

rsocket/src/extension/composite.rs

@@ -119,13 +119,7 @@ impl CompositeMetadata {
         let mime_type = if 0x80 & first != 0 {
             // Well
             let n = first & 0x7F;
-            match MimeType::parse(n) {
-                Some(well) => well,
-                None => {
-                    let err_str = format!("invalid Well-Known MIME type: identifier={:x}", n);
-                    return Err(RSocketError::WithDescription(err_str).into());
-                }
-            }
+            MimeType::WellKnown(n)
         } else {
             // Bad
             let mime_len = (first as usize) + 1;

rsocket/src/extension/mime.rs

@@ -1,8 +1,12 @@
 use std::collections::HashMap;
+use std::convert::TryInto;
 use std::fmt;
 
 use once_cell::sync::Lazy;
 
+use crate::error::RSocketError;
+use crate::Result;
+
 #[derive(PartialEq, Eq, Debug, Clone, Hash)]
 pub enum MimeType {
     Normal(String),
@@ -36,19 +40,26 @@ impl MimeType {
             Self::Normal(_) => None,
         }
     }
-}
 
-impl Into<String> for MimeType {
-    fn into(self) -> String {
-        self.as_ref().to_owned()
+    pub fn as_str(&self) -> Option<&str> {
+        match self {
+            Self::Normal(s) => Some(s.as_ref()),
+            Self::WellKnown(n) => match U8_TO_STR.get(n) {
+                Some(v) => Some(v),
+                None => None,
+            },
+        }
     }
 }
 
-impl AsRef<str> for MimeType {
-    fn as_ref(&self) -> &str {
+impl Into<String> for MimeType {
+    fn into(self) -> String {
         match self {
-            Self::Normal(s) => &s,
-            Self::WellKnown(n) => U8_TO_STR.get(n).unwrap(),
+            Self::Normal(s) => s,
+            Self::WellKnown(n) => match U8_TO_STR.get(&n) {
+                Some(v) => v.to_string(),
+                None => "UNKNOWN".to_string(),
+            },
         }
     }
 }

rsocket/src/payload/setup.rs

@@ -86,12 +86,12 @@ impl SetupPayloadBuilder {
         self
     }
 
-    pub fn set_data_mime_type(mut self, mime: &str) -> Self {
-        self.inner.mime_d = Some(Bytes::from(mime.to_owned()));
+    pub fn set_data_mime_type(mut self, mime: impl Into<String>) -> Self {
+        self.inner.mime_d = Some(Bytes::from(mime.into()));
         self
     }
-    pub fn set_metadata_mime_type(mut self, mime: &str) -> Self {
-        self.inner.mime_m = Some(Bytes::from(mime.to_owned()));
+    pub fn set_metadata_mime_type(mut self, mime: impl Into<String>) -> Self {
+        self.inner.mime_m = Some(Bytes::from(mime.into()));
         self
     }
 

rsocket/src/transport/socket.rs

@@ -69,7 +69,7 @@ impl DuplexSocket {
         socket
     }
 
-    pub(crate) async fn setup(&mut self, setup: SetupPayload) {
+    pub(crate) async fn setup(&mut self, setup: SetupPayload) -> Result<()> {
         let mut bu = frame::Setup::builder(0, 0);
         if let Some(s) = setup.data_mime_type() {
             bu = bu.set_mime_data(s);
@@ -86,7 +86,7 @@ impl DuplexSocket {
         if let Some(b) = m {
             bu = bu.set_metadata(b);
         }
-        self.tx.send(bu.build()).expect("Send setup failed");
+        self.tx.send(bu.build()).map_err(|e| e.into())
     }
 
     #[inline]
@@ -128,7 +128,9 @@ impl DuplexSocket {
                         .set_code(error::ERR_REJECT_SETUP)
                         .set_data(Bytes::from(errmsg))
                         .build();
-                    self.tx.send(sending).expect("Reject setup failed");
+                    if let Err(_) = self.tx.send(sending) {
+                        error!("Reject setup failed");
+                    }
                     return;
                 }
             }
@@ -261,13 +263,28 @@ impl DuplexSocket {
         self.joiners.remove(&sid);
         // pick handler
         if let Some((_, handler)) = self.handlers.remove(&sid) {
-            let desc = input.get_data_utf8().unwrap().to_owned();
+            let desc = input
+                .get_data_utf8()
+                .map(|it| it.to_string())
+                .unwrap_or_default();
             let e = RSocketError::must_new_from_code(input.get_code(), desc);
             match handler {
-                Handler::ReqRR(tx) => tx.send(Err(e.into())).expect("Send RR failed"),
+                Handler::ReqRR(tx) => {
+                    if let Err(_) = tx.send(Err(e.into())) {
+                        error!("respond with error for REQUEST_RESPONSE failed!");
+                    }
+                }
                 Handler::ResRR(_) => unreachable!(),
-                Handler::ReqRS(tx) => tx.send(Err(e.into())).await.expect("Send RS failed"),
-                Handler::ReqRC(tx) => tx.send(Err(e.into())).await.expect("Send RC failed"),
+                Handler::ReqRS(tx) => {
+                    if let Err(_) = tx.send(Err(e.into())).await {
+                        error!("respond with error for REQUEST_STREAM failed!");
+                    };
+                }
+                Handler::ReqRC(tx) => {
+                    if let Err(_) = tx.send(Err(e.into())).await {
+                        error!("respond with error for REQUEST_CHANNEL failed!");
+                    }
+                }
             }
         }
     }
@@ -281,7 +298,9 @@ impl DuplexSocket {
             match handler {
                 Handler::ReqRR(sender) => {
                     info!("REQUEST_RESPONSE {} cancelled!", sid);
-                    sender.send(e).unwrap();
+                    if let Err(_) = sender.send(e) {
+                        error!("notify cancel for REQUEST_RESPONSE failed: sid={}", sid);
+                    }
                 }
                 Handler::ResRR(c) => {
                     let lefts = c.count_down();
@@ -305,21 +324,26 @@ impl DuplexSocket {
                     Handler::ReqRR(_) => match o.remove() {
                         Handler::ReqRR(sender) => {
                             if flag & Frame::FLAG_NEXT != 0 {
-                                sender.send(Ok(Some(input))).unwrap();
+                                if let Err(_) = sender.send(Ok(Some(input))) {
+                                    error!("response successful payload for REQUEST_RESPONSE failed: sid={}",sid);
+                                }
                             } else {
-                                sender.send(Ok(None)).unwrap();
+                                if let Err(_) = sender.send(Ok(None)) {
+                                    error!("response successful payload for REQUEST_RESPONSE failed: sid={}",sid);
+                                }
                             }
                         }
                         _ => unreachable!(),
                     },
                     Handler::ResRR(c) => unreachable!(),
                     Handler::ReqRS(sender) => {
                         if flag & Frame::FLAG_NEXT != 0 {
-                            sender
-                                .clone()
-                                .send(Ok(input))
-                                .await
-                                .expect("Send payload response failed.");
+                            if let Err(e) = sender.send(Ok(input)).await {
+                                error!(
+                                    "response successful payload for REQUEST_STREAM failed: sid={}",
+                                    sid
+                                );
+                            }
                         }
                         if flag & Frame::FLAG_COMPLETE != 0 {
                             o.remove();
@@ -328,11 +352,9 @@ impl DuplexSocket {
                     Handler::ReqRC(sender) => {
                         // TODO: support channel
                         if flag & Frame::FLAG_NEXT != 0 {
-                            sender
-                                .clone()
-                                .send(Ok(input))
-                                .await
-                                .expect("Send payload response failed");
+                            if let Err(_) = sender.clone().send(Ok(input)).await {
+                                error!("response successful payload for REQUEST_CHANNEL failed: sid={}",sid);
+                            }
                         }
                         if flag & Frame::FLAG_COMPLETE != 0 {
                             o.remove();
@@ -396,7 +418,9 @@ impl DuplexSocket {
             }
 
             // async remove canceller
-            canceller.send(sid).await.expect("Send canceller failed");
+            if let Err(_) = canceller.send(sid).await {
+                error!("Send canceller failed: sid={}", sid);
+            }
 
             match result {
                 Ok(Some(res)) => {