hcq/oceanbase
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

[CdcService] Fix Read Amplification in CdcService

Browse Source
This commit is contained in:
zxlzxlzxlzxlzxl
2023-06-09 10:18:09 +00:00
committed by ob-robot
parent 7169c14737
commit 6874d2fbbd
2 changed files with 10 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
src/logservice/cdcservice/ob_cdc_fetcher.cpp
View File

@ -555,6 +555,9 @@ int ObCdcFetcher::ls_fetch_log_(const ObLSID &ls_id,
// exit // exit
reach_max_lsn = true; reach_max_lsn = true;
} }
} else if (OB_NEED_RETRY == ret) {
frt.stop("ArchiveNeedRetry");
ret = OB_SUCCESS;
} else if (OB_ALREADY_IN_NOARCHIVE_MODE == ret || OB_ENTRY_NOT_EXIST == ret) { } else if (OB_ALREADY_IN_NOARCHIVE_MODE == ret || OB_ENTRY_NOT_EXIST == ret) {
// archive is not on or lsn less than the start_lsn in archive // archive is not on or lsn less than the start_lsn in archive
ret = OB_ERR_OUT_OF_LOWER_BOUND; ret = OB_ERR_OUT_OF_LOWER_BOUND;
@ -604,7 +607,9 @@ int ObCdcFetcher::ls_fetch_log_(const ObLSID &ls_id,
} else { } else {
// log fetched successfully // log fetched successfully
fetched_log_count++; fetched_log_count++;
if (resp.log_reach_threshold()) {
frt.stop("LogReachThreshold");
}
LOG_TRACE("LS fetch a log", K(ls_id), K(fetched_log_count), K(frt)); LOG_TRACE("LS fetch a log", K(ls_id), K(fetched_log_count), K(frt));
} }
} }

4
src/logservice/cdcservice/ob_cdc_req.h
View File

@ -427,6 +427,9 @@ public:
pos_ += want_size; pos_ += want_size;
log_num_++; log_num_++;
} }
bool log_reach_threshold() const {
return pos_ > FETCH_BUF_THRESHOLD;
}
bool is_valid() const bool is_valid() const
{ {
return pos_ >= 0 && pos_ <= FETCH_BUF_LEN; return pos_ >= 0 && pos_ <= FETCH_BUF_LEN;
@ -446,6 +449,7 @@ public:
private: private:
static const int64_t FETCH_BUF_LEN = palf::MAX_LOG_BUFFER_SIZE * 8; static const int64_t FETCH_BUF_LEN = palf::MAX_LOG_BUFFER_SIZE * 8;
static const int64_t FETCH_BUF_THRESHOLD = FETCH_BUF_LEN - palf::MAX_LOG_BUFFER_SIZE;
private: private:
int64_t rpc_ver_; int64_t rpc_ver_;