Merge pull request #479 from weibocom/redisproto

redis协议可重入优化

protocol/src/error.rs

@@ -11,7 +11,8 @@ pub enum Error {
     ChanDisabled,
     ChanWriteClosed,
     ChanReadClosed,
-    ProtocolIncomplete,
+    //协议完整至少还需要x个字节
+    ProtocolIncomplete(usize),
     RequestInvalidMagic,
     ResponseInvalidMagic,
     RequestProtocolInvalid,

protocol/src/kv/error.rs

@@ -33,7 +33,7 @@ impl Into<crate::Error> for Error {
                 log::warn!("found unhandle response: {}", packet.utf8());
                 crate::Error::ResponseProtocolInvalid
             }
-            Self::ProtocolIncomplete => crate::Error::ProtocolIncomplete,
+            Self::ProtocolIncomplete => crate::Error::ProtocolIncomplete(0),
         }
     }
 }

protocol/src/kv/mod.rs

@@ -74,7 +74,7 @@ impl Protocol for Kv {
     ) -> crate::Result<HandShake> {
         match self.handshake_inner(stream, option) {
             Ok(h) => Ok(h),
-            Err(crate::Error::ProtocolIncomplete) => Ok(HandShake::Continue),
+            Err(crate::Error::ProtocolIncomplete(0)) => Ok(HandShake::Continue),
             Err(e) => {
                 log::warn!("+++ found err when shake hand:{:?}", e);
                 Err(e)
@@ -143,7 +143,7 @@ impl Protocol for Kv {
         // 解析完毕rsp后，除了数据未读完的场景，其他不管是否遇到err，都要进行take
         match self.parse_response_inner(&mut rsp_packet) {
             Ok(cmd) => Ok(Some(cmd)),
-            Err(crate::Error::ProtocolIncomplete) => Ok(None),
+            Err(crate::Error::ProtocolIncomplete(0)) => Ok(None),
             Err(e) => {
                 // 非MysqlError需要日志并外层断连处理
                 log::error!("+++ err when parse mysql response: {:?}", e);

protocol/src/kv/rsppacket.rs

@@ -196,7 +196,7 @@ impl<'a, S: crate::Stream> ResponsePacket<'a, S> {
     // fn _next_packet(&mut self) -> Result<RingSlice> {
     //     match self.try_next_packet() {
     //         Ok(pld) => Ok(pld),
-    //         Err(Error::ProtocolIncomplete) => Err(crate::Error::ProtocolIncomplete),
+    //         Err(Error::ProtocolIncomplete(0)) => Err(crate::Error::ProtocolIncomplete(0)),
     //         Err(e) => {
     //             // 发现异常，说明异常数据已读完，此处统一take
     //             self.take();

protocol/src/msgque/mcq_bk/text/mcreqpacket.rs

@@ -81,7 +81,7 @@ impl<'a, S: crate::Stream> RequestPacket<'a, S> {
         if self.oft <= self.data.len() {
             return Ok(());
         }
-        return Err(super::Error::ProtocolIncomplete);
+        return Err(super::Error::ProtocolIncomplete(0));
     }
 
     #[inline]
@@ -331,7 +331,7 @@ impl<'a, S: crate::Stream> RequestPacket<'a, S> {
                 ReqPacketState::Val => {
                     m = self.oft + self.vlen;
                     if m >= self.data.len() {
-                        return Err(super::Error::ProtocolIncomplete);
+                        return Err(super::Error::ProtocolIncomplete(0));
                     }
                     if self.data.at(m) == CR {
                         self.skip(self.vlen)?;
@@ -435,7 +435,7 @@ impl<'a, S: crate::Stream> RequestPacket<'a, S> {
                 }
             }
         }
-        Err(super::Error::ProtocolIncomplete)
+        Err(super::Error::ProtocolIncomplete(0))
     }
 
     #[inline]

protocol/src/msgque/mcq_bk/text/mcrsppacket.rs

@@ -53,7 +53,7 @@ impl<'a, S: crate::Stream> RspPacket<'a, S> {
         let mut state = self.state;
 
         if self.data.len() < 2 {
-            return Err(super::Error::ProtocolIncomplete);
+            return Err(super::Error::ProtocolIncomplete(0));
         }
 
         if self.current().is_ascii_digit() {
@@ -252,7 +252,7 @@ impl<'a, S: crate::Stream> RspPacket<'a, S> {
                     m = self.oft + self.vlen;
                     // TODO 提升解析性能 speedup fishermen
                     if m >= self.data.len() {
-                        return Err(super::Error::ProtocolIncomplete);
+                        return Err(super::Error::ProtocolIncomplete(0));
                     }
                     match self.data.at(m) {
                         CR => {
@@ -317,7 +317,7 @@ impl<'a, S: crate::Stream> RspPacket<'a, S> {
 
             self.skip(1)?;
         }
-        Err(super::Error::ProtocolIncomplete)
+        Err(super::Error::ProtocolIncomplete(0))
     }
 
     #[inline]
@@ -340,7 +340,7 @@ impl<'a, S: crate::Stream> RspPacket<'a, S> {
         if self.oft <= self.data.len() {
             return Ok(());
         }
-        return Err(super::Error::ProtocolIncomplete);
+        return Err(super::Error::ProtocolIncomplete(0));
     }
 
     #[inline]

protocol/src/msgque/mcq_bk/text/mod.rs

@@ -97,7 +97,7 @@ impl Protocol for McqText {
     ) -> Result<()> {
         match self.parse_request_inner(stream, alg, process) {
             Ok(_) => Ok(()),
-            Err(Error::ProtocolIncomplete) => Ok(()),
+            Err(Error::ProtocolIncomplete(0)) => Ok(()),
             e => e,
         }
     }
@@ -106,7 +106,7 @@ impl Protocol for McqText {
     fn parse_response<S: Stream>(&self, data: &mut S) -> Result<Option<Command>> {
         match self.parse_response_inner(data) {
             Ok(cmd) => Ok(cmd),
-            Err(Error::ProtocolIncomplete) => Ok(None),
+            Err(Error::ProtocolIncomplete(0)) => Ok(None),
             e => e,
         }
     }

protocol/src/msgque/mcq_bk/text/reqpacket.rs

@@ -53,7 +53,7 @@ impl<'a, S: crate::Stream> RequestPacket<'a, S> {
         if self.oft <= self.data.len() {
             return Ok(());
         }
-        return Err(super::Error::ProtocolIncomplete);
+        return Err(super::Error::ProtocolIncomplete(0));
     }
 
     #[inline]
@@ -229,7 +229,7 @@ impl<'a, S: crate::Stream> RequestPacket<'a, S> {
                 ReqPacketState::Val => {
                     m = self.oft + vlen;
                     if m >= self.data.len() {
-                        return Err(super::Error::ProtocolIncomplete);
+                        return Err(super::Error::ProtocolIncomplete(0));
                     }
                     if self.data.at(m) == CR {
                         self.skip(vlen)?;
@@ -251,7 +251,7 @@ impl<'a, S: crate::Stream> RequestPacket<'a, S> {
             // 当前字节处理完毕，继续下一个字节
             self.skip(1)?;
         }
-        Err(super::Error::ProtocolIncomplete)
+        Err(super::Error::ProtocolIncomplete(0))
     }
 
     #[inline]
@@ -324,7 +324,7 @@ impl Packet for RingSlice {
             *oft = idx + 2;
             Ok(())
         } else {
-            Err(crate::Error::ProtocolIncomplete)
+            Err(crate::Error::ProtocolIncomplete(0))
         }
     }
 
@@ -351,7 +351,7 @@ impl Packet for RingSlice {
                 return Ok(());
             }
         }
-        Err(super::Error::ProtocolIncomplete)
+        Err(super::Error::ProtocolIncomplete(0))
     }
 }
 // mcq 解析时状态

protocol/src/msgque/mcq_bk/text/rsppacket.rs

@@ -55,7 +55,7 @@ impl<'a, S: crate::Stream> RspPacket<'a, S> {
     #[inline]
     fn start_with(&self, oft: usize, s: &[u8]) -> Result<bool> {
         if oft + s.len() > self.data.len() {
-            Err(crate::Error::ProtocolIncomplete)
+            Err(crate::Error::ProtocolIncomplete(0))
         } else {
             Ok(self.data.start_with(oft, s))
         }
@@ -64,7 +64,7 @@ impl<'a, S: crate::Stream> RspPacket<'a, S> {
     // memcache rsponse 解析的状态机，后续考虑优化 fishermen
     pub(super) fn parse(&mut self) -> Result<()> {
         if self.data.len() < 2 {
-            return Err(super::Error::ProtocolIncomplete);
+            return Err(super::Error::ProtocolIncomplete(0));
         }
 
         let mut state = RspPacketState::RspStr;
@@ -209,7 +209,7 @@ impl<'a, S: crate::Stream> RspPacket<'a, S> {
                 RspPacketState::Val => {
                     token = self.oft + vlen;
                     if token >= self.data.len() {
-                        return Err(super::Error::ProtocolIncomplete);
+                        return Err(super::Error::ProtocolIncomplete(0));
                     }
                     self.skip(vlen)?;
                     match self.current() {
@@ -276,7 +276,7 @@ impl<'a, S: crate::Stream> RspPacket<'a, S> {
 
             self.skip(1)?;
         }
-        Err(super::Error::ProtocolIncomplete)
+        Err(super::Error::ProtocolIncomplete(0))
     }
 
     pub(super) fn delay_metric(&mut self) -> Result<()> {
@@ -344,7 +344,7 @@ impl<'a, S: crate::Stream> RspPacket<'a, S> {
         if self.oft <= self.data.len() {
             return Ok(());
         }
-        return Err(super::Error::ProtocolIncomplete);
+        return Err(super::Error::ProtocolIncomplete(0));
     }
 
     #[inline]

protocol/src/redis/command.rs

@@ -56,7 +56,7 @@ impl CommandHasher {
             }
             h.hash(slice[i]);
         }
-        Err(crate::Error::ProtocolIncomplete)
+        Err(crate::Error::ProtocolIncomplete(0))
     }
 }
 

protocol/src/redis/mod.rs

@@ -70,32 +70,23 @@ impl Redis {
     }
 
     #[inline]
-    fn parse_response_inner<S: Stream>(&self, s: &mut S) -> Result<Option<Command>> {
+    pub fn parse_response_inner<S: Stream>(&self, s: &mut S) -> Result<Option<Command>> {
         let data: Packet = s.slice().into();
         let ctx: &mut ResponseContext = transmute(s.context());
         log::debug!("+++ will parse redis rsp:{:?}", data);
-        // data.check_onetoken(*oft)?;
 
         match data.at(0) {
             b'-' | b':' | b'+' => data.line(&mut ctx.oft)?,
-            b'$' => ctx.oft += data.num_of_string(&mut ctx.oft)? + 2,
-            b'*' => data.skip_multibulks(ctx)?,
+            b'$' => data.skip_string_check(&mut ctx.oft)?,
+            b'*' => data.skip_multibulks_with_ctx(ctx)?,
             _ => return Err(RedisError::RespInvalid.into()),
         }
 
         let oft = ctx.oft;
-        ctx.oft = 0; // 响应消息是b'$'，若数据未接收完整，下次需要从起始位置开始解析
-        match oft <= data.len() {
-            true => Ok(Some(Command::from_ok(s.take(oft)))),
-            false => Err(Error::ProtocolIncomplete),
-        }
-        // Ok((*oft <= data.len()).then(|| Command::from_ok(s.take(*oft))))
-    }
-    #[inline(always)]
-    fn left_bytes<S: Stream>(&self, s: &mut S) -> usize {
-        let ctx = transmute(s.context());
-        // 64是经验值
-        ctx.bulk * 64
+        assert!(oft != 0);
+        assert!(oft <= data.len());
+        ctx.oft = 0;
+        Ok(Some(Command::from_ok(s.take(oft))))
     }
 }
 
@@ -117,7 +108,7 @@ impl Protocol for Redis {
         let mut packet = RequestPacket::new(stream);
         match self.parse_request_inner(&mut packet, alg, process) {
             Ok(_) => Ok(()),
-            Err(Error::ProtocolIncomplete) => {
+            Err(Error::ProtocolIncomplete(0)) => {
                 // 如果解析数据不够，提前reserve stream的空间
                 packet.reserve_stream_buff();
                 Ok(())
@@ -134,18 +125,10 @@ impl Protocol for Redis {
     fn parse_response<S: Stream>(&self, data: &mut S) -> Result<Option<Command>> {
         match self.parse_response_inner(data) {
             Ok(cmd) => Ok(cmd),
-            Err(Error::ProtocolIncomplete) => {
-                let ctx = transmute(data.context());
-                let oft = ctx.oft;
-                //assert!(oft + 3 >= data.len(), "oft:{} => {:?}", oft, data.slice());
-                if ctx.bulk > 0 {
-                    // 响应消息是array场景
-                    let left = self.left_bytes(data);
+            Err(Error::ProtocolIncomplete(left)) => {
+                if left > 0 {
                     data.reserve(left);
-                } else if oft > data.len() {
-                    data.reserve(oft - data.len());
                 }
-
                 Ok(None)
             }
             e => e,