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

Move ls_id reference usage in archive and restore

Browse Source
This commit is contained in:
obdev
2023-03-28 02:11:53 +00:00
committed by ob-robot
parent 399b29e4ba
commit 1bb857cd7e
6 changed files with 15 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
src/logservice/archiveservice/ob_archive_fetcher.cpp
View File

@ -322,7 +322,7 @@ int ObArchiveFetcher::handle_single_task_()
ARCHIVE_LOG(ERROR, "data is NULL", K(ret), K(data)); ARCHIVE_LOG(ERROR, "data is NULL", K(ret), K(data));
} else { } else {
ObArchiveLogFetchTask *task = static_cast<ObArchiveLogFetchTask*>(data); ObArchiveLogFetchTask *task = static_cast<ObArchiveLogFetchTask*>(data);
ObLSID id = task->get_ls_id_copy(); ObLSID id = task->get_ls_id();
ArchiveKey key = task->get_station().get_round(); ArchiveKey key = task->get_station().get_round();
// task will be submit to fetch_log_queue or re-submit to handle or free due to fatal error // task will be submit to fetch_log_queue or re-submit to handle or free due to fatal error
@ -511,7 +511,7 @@ int ObArchiveFetcher::init_helper_(ObArchiveLogFetchTask &task, const LSN &commi
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
LSN start_offset; LSN start_offset;
const ObLSID &id = task.get_ls_id(); const ObLSID id = task.get_ls_id();
const LSN &end_offset = task.get_end_offset(); const LSN &end_offset = task.get_end_offset();
const ObArchivePiece &cur_piece = task.get_piece(); const ObArchivePiece &cur_piece = task.get_piece();
const ObArchivePiece &next_piece = task.get_next_piece(); const ObArchivePiece &next_piece = task.get_next_piece();
@ -771,7 +771,7 @@ int ObArchiveFetcher::update_log_fetch_task_(ObArchiveLogFetchTask &fetch_task,
int ObArchiveFetcher::submit_fetch_log_(ObArchiveLogFetchTask &task, bool &submitted) int ObArchiveFetcher::submit_fetch_log_(ObArchiveLogFetchTask &task, bool &submitted)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
const ObLSID &id = task.get_ls_id(); const ObLSID id = task.get_ls_id();
submitted = false; submitted = false;
if (! task.has_fetch_log()) { if (! task.has_fetch_log()) {
@ -779,10 +779,10 @@ int ObArchiveFetcher::submit_fetch_log_(ObArchiveLogFetchTask &task, bool &submi
} else { } else {
GET_LS_TASK_CTX(ls_mgr_, id) { GET_LS_TASK_CTX(ls_mgr_, id) {
if (OB_FAIL(ls_archive_task->push_fetch_log(task))) { if (OB_FAIL(ls_archive_task->push_fetch_log(task))) {
ARCHIVE_LOG(WARN, "push fetch log failed", K(ret), K(task)); ARCHIVE_LOG(WARN, "push fetch log failed", K(ret), K(id), K(task));
} else { } else {
submitted = true; submitted = true;
ARCHIVE_LOG(INFO, "push fetch log succ", KP(&task)); ARCHIVE_LOG(INFO, "push fetch log succ", K(id), KP(&task));
} }
} }
} }

8
src/logservice/archiveservice/ob_archive_sender.cpp
View File

@ -217,14 +217,14 @@ int ObArchiveSender::modify_thread_count(const int64_t thread_count)
int ObArchiveSender::submit_send_task_(ObArchiveSendTask *task) int ObArchiveSender::submit_send_task_(ObArchiveSendTask *task)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
const ObLSID &id = task->get_ls_id(); const ObLSID id = task->get_ls_id();
if (OB_ISNULL(ls_mgr_)) { if (OB_ISNULL(ls_mgr_)) {
ret = OB_ERR_UNEXPECTED; ret = OB_ERR_UNEXPECTED;
ARCHIVE_LOG(ERROR, "ls_mgr_ is NULL", K(ret), K(ls_mgr_)); ARCHIVE_LOG(ERROR, "ls_mgr_ is NULL", K(ret), K(ls_mgr_));
} else { } else {
GET_LS_TASK_CTX(ls_mgr_, id) { GET_LS_TASK_CTX(ls_mgr_, id) {
if (OB_FAIL(ls_archive_task->push_send_task(*task, *this))) { if (OB_FAIL(ls_archive_task->push_send_task(*task, *this))) {
ARCHIVE_LOG(WARN, "push_send_task fail", K(ret), KPC(task)); ARCHIVE_LOG(WARN, "push_send_task fail", K(ret), K(id), KPC(task));
} }
} }
} }
@ -428,7 +428,7 @@ bool ObArchiveSender::in_normal_status_(const ArchiveKey &key) const
void ObArchiveSender::handle(ObArchiveSendTask &task, TaskConsumeStatus &consume_status) void ObArchiveSender::handle(ObArchiveSendTask &task, TaskConsumeStatus &consume_status)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
const ObLSID &id = task.get_ls_id(); const ObLSID id = task.get_ls_id();
const ArchiveWorkStation &station = task.get_station(); const ArchiveWorkStation &station = task.get_station();
share::ObBackupDest backup_dest; share::ObBackupDest backup_dest;
if (OB_UNLIKELY(! task.is_valid())) { if (OB_UNLIKELY(! task.is_valid())) {
@ -507,7 +507,7 @@ int ObArchiveSender::check_piece_continuous_(const ObArchiveSendTask &task,
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
ObLSArchivePersistInfo info; ObLSArchivePersistInfo info;
const ObLSID &id = task.get_ls_id(); const ObLSID id = task.get_ls_id();
const ObArchivePiece &piece = task.get_piece(); const ObArchivePiece &piece = task.get_piece();
const ArchiveWorkStation &station = task.get_station(); const ArchiveWorkStation &station = task.get_station();
if (! ls_task_tuple.get_piece().is_valid()) { if (! ls_task_tuple.get_piece().is_valid()) {

5
src/logservice/archiveservice/ob_archive_task.h
View File

@ -71,8 +71,7 @@ public:
const LSN &start_lsn, const LSN &start_lsn,
const LSN &end_lsn); const LSN &end_lsn);
uint64_t get_tenant_id() const { return tenant_id_; } uint64_t get_tenant_id() const { return tenant_id_; }
const ObLSID &get_ls_id() const { return id_; } ObLSID get_ls_id() const { return id_; }
ObLSID get_ls_id_copy() { return id_; }
const ArchiveWorkStation &get_station() { return station_; } const ArchiveWorkStation &get_station() { return station_; }
const LSN &get_start_offset() const { return start_offset_; } const LSN &get_start_offset() const { return start_offset_; }
const LSN &get_cur_offset() const { return cur_offset_; } const LSN &get_cur_offset() const { return cur_offset_; }
@ -163,7 +162,7 @@ public:
const share::SCN &max_scn); const share::SCN &max_scn);
bool is_valid() const; bool is_valid() const;
uint64_t get_tenant_id() const { return tenant_id_;} uint64_t get_tenant_id() const { return tenant_id_;}
const ObLSID &get_ls_id() const { return id_; } ObLSID get_ls_id() const { return id_; }
const ArchiveWorkStation &get_station() const { return station_; } const ArchiveWorkStation &get_station() const { return station_; }
const ObArchivePiece &get_piece() const { return piece_; } const ObArchivePiece &get_piece() const { return piece_; }
const LSN &get_start_lsn() const { return start_offset_; } const LSN &get_start_lsn() const { return start_offset_; }

2
src/logservice/archiveservice/ob_ls_mgr.cpp
View File

@ -479,7 +479,7 @@ int ObArchiveLSMgr::add_task_(const ObLSID &id,
int ObArchiveLSMgr::insert_or_update_ls_(const StartArchiveHelper &helper) int ObArchiveLSMgr::insert_or_update_ls_(const StartArchiveHelper &helper)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
const ObLSID &id = helper.get_ls_id(); const ObLSID id = helper.get_ls_id();
if (OB_UNLIKELY(! helper.is_valid())) { if (OB_UNLIKELY(! helper.is_valid())) {
ARCHIVE_LOG(WARN, "helper is not valid", KR(ret), K(helper)); ARCHIVE_LOG(WARN, "helper is not valid", KR(ret), K(helper));

2
src/logservice/archiveservice/ob_ls_task.cpp
View File

@ -713,7 +713,7 @@ int ObLSArchiveTask::ArchiveDest::push_fetch_log(ObArchiveLogFetchTask &task)
int ObLSArchiveTask::ArchiveDest::push_send_task(ObArchiveSendTask &task, ObArchiveWorker &worker) int ObLSArchiveTask::ArchiveDest::push_send_task(ObArchiveSendTask &task, ObArchiveWorker &worker)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
const ObLSID &id = task.get_ls_id(); const ObLSID id = task.get_ls_id();
if (NULL == send_task_queue_) { if (NULL == send_task_queue_) {
if (OB_ISNULL(send_task_queue_ = allocator_->alloc_send_task_status(id))) { if (OB_ISNULL(send_task_queue_ = allocator_->alloc_send_task_status(id))) {
ret = OB_ALLOCATE_MEMORY_FAILED; ret = OB_ALLOCATE_MEMORY_FAILED;

4
src/logservice/restoreservice/ob_remote_fetch_log_worker.cpp
View File

@ -337,7 +337,7 @@ int ObRemoteFetchWorker::submit_entries_(ObFetchLogTask &task)
const char *buf = NULL; const char *buf = NULL;
int64_t size = 0; int64_t size = 0;
LSN lsn; LSN lsn;
const ObLSID &id = task.id_; const ObLSID id = task.id_;
while (OB_SUCC(ret) && ! has_set_stop()) { while (OB_SUCC(ret) && ! has_set_stop()) {
bool quota_done = false; bool quota_done = false;
if (OB_FAIL(task.iter_.next(entry, lsn, buf, size))) { if (OB_FAIL(task.iter_.next(entry, lsn, buf, size))) {
@ -447,7 +447,7 @@ int ObRemoteFetchWorker::try_retire_(ObFetchLogTask *&task)
int ObRemoteFetchWorker::push_submit_array_(ObFetchLogTask &task) int ObRemoteFetchWorker::push_submit_array_(ObFetchLogTask &task)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
const ObLSID &id = task.id_; const ObLSID id = task.id_;
DEBUG_SYNC(BEFORE_RESTORE_SERVICE_PUSH_FETCH_DATA); DEBUG_SYNC(BEFORE_RESTORE_SERVICE_PUSH_FETCH_DATA);
GET_RESTORE_HANDLER_CTX(id) { GET_RESTORE_HANDLER_CTX(id) {
if (OB_FAIL(restore_handler->submit_sorted_task(task))) { if (OB_FAIL(restore_handler->submit_sorted_task(task))) {