Merge pull request #368 from weibocom/dev_update_kv_error

Dev update kv error & 支持crc32abs
weibocom · Oct 9, 2023 · 9d08d2d · 9d08d2d
2 parents 399e7a0 + 27e958e
commit 9d08d2d
Show file tree

Hide file tree

Showing 5 changed files with 63 additions and 6 deletions.
diff --git a/protocol/src/kv/error.rs b/protocol/src/kv/error.rs
@@ -1,5 +1,7 @@
 use std::{fmt::Display, io::ErrorKind};
 
+use ds::Utf8;
+
 // /// 按照mc标准error进行设置异常内容
 // // pub(super) const REQ_INVALID: &str = "CLIENT_ERROR request is invalid";
 pub(super) const REQ_INVALID_KEY: &str = "CLIENT_ERROR request key is invalid";
@@ -27,8 +29,9 @@ impl Into<crate::Error> for Error {
             Self::RequestInvalid(packet) => crate::Error::FlushOnClose(packet.into()),
             Self::RequestInvalidKey(packet) => crate::Error::FlushOnClose(packet.into()),
             Self::UnhandleResponseError(packet) => {
-                // 该异常需要构建成response，不能转为error传出
-                panic!("kv unhanlde rsp err: {:?}", packet);
+                // auth时，如果有这种异常，需要上抛异常断连接
+                log::warn!("found unhandle response: {}", packet.utf8());
+                crate::Error::ResponseProtocolInvalid
             }
             Self::ProtocolIncomplete => crate::Error::ProtocolIncomplete,
         }

diff --git a/protocol/src/kv/rsppacket.rs b/protocol/src/kv/rsppacket.rs
@@ -233,7 +233,8 @@ impl<'a, S: crate::Stream> ResponsePacket<'a, S> {
                 match ParseBuf::from(self.data.sub_slice(self.oft, left_len))
                     .parse(self.capability_flags())?
                 {
-                    // server返回的异常响应转为MysqlError，最终传给client，不用断连接
+                    // server返回的异常响应转为MysqlError，对于client request，最终传给client，不用断连接；
+                    // 对于连接初期的auth，则直接断连接即可；
                     ErrPacket::Error(server_error) => {
                         // self.handle_err();
                         self.oft += header.payload_len;

diff --git a/sharding/src/hash/crc32.rs b/sharding/src/hash/crc32.rs
@@ -1,6 +1,20 @@
-use super::DebugName;
+use super::{DebugName, Hash};
 use std::fmt::Display;
 
+/// 当前crc32相关hash算法目前有两类，基于i64提升的非负crc32，基于i32的crc32abs:：
+///   ================  基于i64版本  ================
+///   1 crc32: java版本的i64/long版本的crc32，对整个key做crc32算法；
+///   2 crc32-num: 对偏移N个字节后的数字进行crc32计算, eg: crc32-num-5, crc32-num；
+///   3 crc32-short: 对crc32计算后，再转为short，适配java mc访问；
+///   4 crc32-delimiter: 目前有3种，crc32-point, crc32-pound, crc32-underscore，表示对特殊字符后的内容做crc32，后面可以再跟一个数字表示偏移；
+///   5 crc32-mixnum:key中所有num拼接成一个字串num做hashkey，like a_123_456_bc的hashkey是123456；
+///   6 crc32-smartnum: 第一串长度大于等于5的数字做为hashkey;
+///   ================  基于i32版本  ================
+///     备注：不取名crc32-abs，是为了区分i32/i63的base，及为后续扩展做准备 fishermen
+///   7 crc32abs: 转为i32，然后进行abs操作
+///   8 后续可能会有crc32abs-xxx，此处先预留语义；
+///
+
 pub(super) const CRC32TAB: [i64; 256] = [
     0x00000000, 0x77073096, 0xEE0E612C, 0x990951BA, 0x076DC419, 0x706AF48F, 0xE963A535, 0x9E6495A3,
     0x0EDB8832, 0x79DCB8A4, 0xE0D5E91E, 0x97D2D988, 0x09B64C2B, 0x7EB17CBD, 0xE7B82D07, 0x90BF1D91,
@@ -82,7 +96,8 @@ impl super::Hash for Crc32 {
         crc ^= CRC_SEED;
         crc &= CRC_SEED;
         if crc <= 0 {
-            log::debug!("crc32 - error hash/{} for key/{:?}", crc, key);
+            // 理论上不会有负数hash
+            log::warn!("crc32 - negative hash/{} for key/{:?}", crc, key);
         }
         crc
     }
@@ -315,3 +330,29 @@ impl super::Hash for Crc32MixNum {
         crc
     }
 }
+
+/// 遵从i32进行crc32计算，并进行abs操作;
+#[derive(Debug, Clone, Default)]
+pub struct Crc32Abs;
+
+impl Hash for Crc32Abs {
+    fn hash<S: super::HashKey>(&self, key: &S) -> i64 {
+        let mut crc: i64 = CRC_SEED;
+
+        for i in 0..key.len() {
+            let c = key.at(i);
+            crc = ((crc >> 8) & 0x00FFFFFF) ^ CRC32TAB[((crc ^ (c as i64)) & 0xff) as usize];
+        }
+
+        crc ^= CRC_SEED;
+        crc &= CRC_SEED;
+
+        let mut crc = crc as i32;
+        if crc <= 0 {
+            // 理论上会有大量负数hash
+            log::debug!("crc32abs - negative hash/{} for key/{:?}", crc, key);
+            crc = crc.abs();
+        }
+        crc as i64
+    }
+}
diff --git a/sharding/src/hash/mod.rs b/sharding/src/hash/mod.rs
@@ -76,7 +76,8 @@ pub enum Hasher {
     Crc32localSmartNum(Crc32localSmartNum), //crc32 for key like： xxx + id + xxx，id的长度需要大于等于5
     LBCrc32localDelimiter(LBCrc32localDelimiter), // long bytes crc32local for hash like: 123.a, 124_a, 123#a
     Rawcrc32local(Rawcrc32local),                 // raw or crc32local
-    Random(RandomHash),                           // random hash
+    Crc32Abs(Crc32Abs), // crc32abs: 基于i32转换，然后直接取abs；其他走i64提升为正数
+    Random(RandomHash), // random hash
     RawSuffix(RawSuffix),
 }
 
@@ -118,6 +119,7 @@ impl Hasher {
                 "lbcrc32local" => {
                     Self::LBCrc32localDelimiter(LBCrc32localDelimiter::from(alg_lower.as_str()))
                 }
+                "crc32abs" => Self::Crc32Abs(Default::default()),
                 "random" => Self::Random(Default::default()),
                 _ => {
                     // 默认采用mc的crc32-s hash

diff --git a/tests/src/hash_test.rs b/tests/src/hash_test.rs
@@ -73,4 +73,14 @@ mod hash_test {
         println!("key:{}, hash:{}", key2, hash2);
         assert_eq!(hash2, 1108486745);
     }
+
+    #[test]
+    fn crc32abs() {
+        let hasher = Hasher::from("crc32abs");
+        let key = "h14243dc752b5beac".to_string();
+        let crc = hasher.hash(&key.as_bytes());
+        println!("crc: {}, key:{} ", crc, key);
+        // h14243dc752b5beac 对应i64算法为2461123049，i32算法为-1833844247，abs后为1833844247，
+        assert_eq!(crc, 1833844247);
+    }
 }