基于C++的京东自研分布式表缓存DTC设计源码

/* * Copyright [2021] JD.com, Inc. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ #include <inttypes.h> #include <math.h> #include <inttypes.h> #include "da_protocal.h" #include "../da_msg.h" #include "../da_conn.h" #include "../da_request.h" #include "../da_buf.h" #include "../da_util.h" #include "../da_errno.h" #include "../da_time.h" #include "../da_core.h" #define HEADER_SIZE sizeof(struct CPacketHeader) #define MAXPACKETSIZE (64<<20) #define MultiKeyValue 32 #define __FLTFMT__ "%LA" #define CONVERT_NULL_TO_EMPTY_STRING 1 enum fieldtype { None = 0, Signed = 1, Unsigned = 2, Float = 3, String = 4, Binary = 5, TotalType, }; enum codefield { FD_HEADER = 0, FD_VERSION = 1, FD_TABLEDEFINE = 2, FD_REQUEST = 3, FD_RESULTINFO = 4, FD_UPDATEINFO = 5, FD_CONDITIONINFO = 6, FD_FIELDSET = 7, FD_RESULTSET = 8, FD_TOTALFIELD = 9, }; enum codestate { ST_ID = 0, ST_LENTH = 1, ST_VALUE = 2, }; enum fieldtype versionInfoDefinition[19] = { None, String, // 1 -- table name Binary, // 2 -- data table hash [for hb register use, unless always send 0x00000...] Unsigned, // 3 -- serial# Binary, // 4 -- real table hash String, // 5 -- client version String, // 6 -- ctlib version String, // 7 -- helper version [unuse] Unsigned, // 8 -- keepalive timeout Unsigned, // 9 -- key type Unsigned, // 10 -- key field count Unsigned, // 11 -- key value count Binary, // 12 -- key type list Binary, // 13 -- key name list Unsigned, // 14 -- hot backup ID Signed, // 15 -- hot backup master timestamp Signed, // 16 -- hot backup slave timestamp Unsigned, // 17 -- agent client id String, // 18 -- accessKey }; enum fieldtype requestInfoDefinition[8] = { Binary, // 0 -- key Unsigned, // 1 -- timeout Unsigned, // Limit start Unsigned, // Limit count String, // 4 -- trace msg Unsigned, // 5 -- Cache ID -- OBSOLETED String, // 6 -- raw config string Unsigned, // 7 -- admin cmd code }; enum fieldtype resultinfoDefinition[10] = { Binary, // 0 -- key, original key or INSERT_ID Signed, // 1 -- code String, // 2 -- trace msg String, // 3 -- from Signed, // 4 -- affected rows, by delete/update Unsigned, // 5 -- total rows, not used. for SELECT SQL_CALC_FOUND_ROWS... Unsigned, // 6 -- insert_id, not used. for SELECT SQL_CALC_FOUND_ROWS... Binary, // 7 -- query server info: version, binlogid,.... Unsigned, // 8 -- server timestamp Unsigned, // 9 -- Hit Flag }; static int dtc_decode_value(enum fieldtype type, int lenth, uint8_t *p, CValue *val) { uint8_t *q; switch (type) { case None: break; case Signed: case Unsigned: if (lenth == 0 || lenth > 8) { goto decode_value_error; } q = (uint8_t *) p + 1; int64_t s64; s64 = *(int8_t *) p; switch (lenth) { case 8: s64 = (s64 << 8) | *q++; case 7: s64 = (s64 << 8) | *q++; case 6: s64 = (s64 << 8) | *q++; case 5: s64 = (s64 << 8) | *q++; case 4: s64 = (s64 << 8) | *q++; case 3: s64 = (s64 << 8) | *q++; case 2: s64 = (s64 << 8) | *q++; } val->s64 = s64; break; case Float: if (lenth < 3) goto decode_value_error; if (p[lenth - 1] != '\0') goto decode_value_error; if (!strcmp((char *) p, "NAN")) val->flt = NAN; else if (!strcmp((char *) p, "INF")) val->flt = INFINITY; else if (!strcmp((char *) p, "-INF")) val->flt = -INFINITY; else { long double ldf; if (sscanf((char *) p, __FLTFMT__, &ldf) != 1) goto decode_value_error; val->flt = ldf; } break; case String: case Binary: if (lenth == 0) { #if CONVERT_NULL_TO_EMPTY_STRING val->str.data = p; val->str.len = 0; #else val->str.data=NULL; val->str.len=0; #endif } else { if (p[lenth - 1] != '\0') goto decode_value_error; val->str.data = p; val->str.len = lenth - 1; } break; default: goto decode_value_error; } return 0; decode_value_error: return -1; } static int dtc_decode_lenth(uint8_t **p) { int len = **p; *p += 1; if (len < 240) { } else if (len <= 252) { len = ((len & 0xF) << 8) + **p; *p += 1; } else if (len == 253) { len = ((**p) << 8) + (*(*p + 1)); *p = *p + 2; } else if (len == 254) { len = ((**p) << 16) + (*(*p + 1) << 8) + (*(*p + 2)); *p = *p + 3; } else { len = ((**p) << 24) + (*(*p + 1) << 16) + (*(*p + 2) << 8) + (*(*p + 3)); *p = *p + 4; if (len > MAXPACKETSIZE) { return -1; } } return len; } static int next_field(struct msg *r, int field) { int i; for (i = field; i < FD_TOTALFIELD - 1; i++) { if (r->seclen[i] != 0) { return i + 1; } } return FD_TOTALFIELD; } /* * parse request msg */ void dtc_parse_req(struct msg *r) { struct mbuf *b; uint8_t *p; int ret; int state; int field; field = r->field; state = r->state; b = STAILQ_LAST(&r->buf_q, mbuf, next); p = r->pos; while (p < b->last) { switch (field) { case FD_HEADER: { r->token = p; if (b->last - p < HEADER_SIZE) { log_debug("receive size small than package header!"); p = b->last; break; } struct CPacketHeader *pheader; pheader = (struct CPacketHeader *) p; r->seclen = pheader->len; r->cmd = (msg_type_t) pheader->cmd; r->flags = pheader->flags; log_debug("seclen:%d,%d,%d,%d,%d,%d,%d,%d", r->seclen[0], r->seclen[1], r->seclen[2], r->seclen[3], r->seclen[4], r->seclen[5], r->seclen[6], r->seclen[7]); p = p + HEADER_SIZE; field = next_field(r, FD_HEADER); if (field == FD_TOTALFIELD) { log_debug("all fields has been parsed,parse done!"); goto success; } log_debug("next field to be parse:%d", field); state = ST_ID; r->sec_parsed_len = 0; r->parsed_len += HEADER_SIZE; r->token = NULL; break; } case FD_VERSION: { r->token = p; switch (state) { case ST_ID: { r->cur_parse_id = *(uint8_t *) p++; if (r->cur_parse_id > 18) { log_error("id:%d,version hasn't slot id max than 18", r->cur_parse_id); goto error; } //record len from start to keycount if (r->cur_parse_id == 11) { r->keycountstartlen = r->parsed_len; } r->cur_parse_type = versionInfoDefinition[r->cur_parse_id]; ++r->sec_parsed_len; ++r->parsed_len; state = ST_LENTH; r->token = NULL; break; } case ST_LENTH: { int len = dtc_decode_lenth(&p); if (p >= b->last) { p = b->last; break; } if (r->cur_parse_id == 11) { r->keycountlen += (p - r->token) + len + 1; } r->cur_parse_lenth = len; r->parsed_len += p - r->token; r->sec_parsed_len += p - r->token; r->token = NULL; state = ST_VALUE; break; } case ST_VALUE: { CValue tempval = { 0 }; switch (r->cur_parse_id) { case 3: { if (b->last - p < r->cur_parse_lenth) { log_debug( "parse version info serialnr id:%d,len:%d,not in one buff!", r->cur_parse_id, r->cur_parse_lenth); p = b->last; break; } ret = dtc_decode_value(r->cur_parse_type, r->cur_parse_lenth, p, &tempval); if (ret < 0) { log_error( "decode version info serialnr id:%d,length:%d error", r->cur_parse_id, r->cur_parse_lenth); goto error; } r->serialnr = tempval.u64; log_debug( "parse version info id:%d success,serialnr: %"PRIu64"", r->cur_parse_id, r->serialnr); p += r->cur_parse_lenth; r->sec_parsed_len += r->cur_parse_lenth; r->parsed_len += r->cur_parse_lenth; r->to