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

migration should use ha_mid_thread_score instead of ha_high_thread_score

Browse Source
This commit is contained in:
oceanoverflow 2024-05-22 11:36:13 +00:00 committed by ob-robot
parent 0e70853366
commit 8e10ce3c36
6 changed files with 169 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
src/share/scheduler/ob_tenant_dag_scheduler.h
View File

@ -356,7 +356,7 @@ public:
bool is_dag_failed() const { return ObIDag::DAG_STATUS_NODE_FAILED == dag_status_; }
void set_add_time() { add_time_ = ObTimeUtility::fast_current_time(); }
int64_t get_add_time() const { return add_time_; }
int64_t get_priority() const { return priority_; }
ObDagPrio::ObDagPrioEnum get_priority() const { return priority_; }
void set_priority(ObDagPrio::ObDagPrioEnum prio) { priority_ = prio; }
const ObDagId &get_dag_id() const { return id_; }
int set_dag_id(const ObDagId &dag_id);
@ -1062,6 +1062,8 @@ public:
template<typename T>
int alloc_dag(T *&dag);
template<typename T>
int alloc_dag_with_priority(const ObDagPrio::ObDagPrioEnum &prio, T *&dag);
template<typename T>
int create_and_add_dag_net(const ObIDagInitParam *param);
void free_dag(ObIDag &dag);
void inner_free_dag(ObIDag &dag);
@ -1297,6 +1299,27 @@ int ObTenantDagScheduler::alloc_dag(T *&dag)
return ret;
}
template <typename T>
int ObTenantDagScheduler::alloc_dag_with_priority(
const ObDagPrio::ObDagPrioEnum &prio, T *&dag)
{
int ret = OB_SUCCESS;
dag = NULL;
if (prio < ObDagPrio::DAG_PRIO_COMPACTION_HIGH
|| prio >= ObDagPrio::DAG_PRIO_MAX) {
ret = OB_INVALID_ARGUMENT;
COMMON_LOG(WARN, "get invalid arg", K(ret), K(prio));
} else if (OB_FAIL(alloc_dag(dag))) {
COMMON_LOG(WARN, "failed to alloc dag", K(ret));
} else if (OB_ISNULL(dag)) {
ret = OB_ERR_UNEXPECTED;
COMMON_LOG(WARN, "dag should not be null", K(ret), KP(dag));
} else {
dag->set_priority(prio);
}
return ret;
}
template<typename T>
void ObTenantDagScheduler::free_dag_net(T *&dag_net)
{

34
src/storage/high_availability/ob_ls_complete_migration.cpp
View File

@ -177,6 +177,7 @@ int ObLSCompleteMigrationDagNet::start_running_for_migration_()
int tmp_ret = OB_SUCCESS;
ObInitialCompleteMigrationDag *initial_dag = nullptr;
ObTenantDagScheduler *scheduler = nullptr;
ObDagPrio::ObDagPrioEnum prio = ObDagPrio::DAG_PRIO_MAX;
if (!is_inited_) {
ret = OB_NOT_INIT;
@ -185,8 +186,10 @@ int ObLSCompleteMigrationDagNet::start_running_for_migration_()
} else if (OB_ISNULL(scheduler = MTL(ObTenantDagScheduler*))) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("failed to get ObTenantDagScheduler from MTL", K(ret));
} else if (OB_FAIL(scheduler->alloc_dag(initial_dag))) {
LOG_WARN("failed to alloc initial dag ", K(ret));
} else if (OB_FAIL(ObMigrationUtils::get_dag_priority(ctx_.arg_.type_, prio))) {
LOG_WARN("failed to get dag priority", K(ret));
} else if (OB_FAIL(scheduler->alloc_dag_with_priority(prio, initial_dag))) {
LOG_WARN("failed to alloc initial dag ", K(ret), K(prio));
} else if (OB_FAIL(initial_dag->init(this))) {
LOG_WARN("failed to init initial dag", K(ret));
} else if (OB_FAIL(add_dag_into_dag_net(*initial_dag))) {
@ -663,8 +666,9 @@ int ObInitialCompleteMigrationDag::fill_dag_key(char *buf, const int64_t buf_len
LOG_WARN("ha dag net ctx type is unexpected", K(ret), KPC(ha_dag_net_ctx_));
} else if (FALSE_IT(self_ctx = static_cast<ObLSCompleteMigrationCtx *>(ha_dag_net_ctx_))) {
} else if (OB_FAIL(databuff_printf(buf, buf_len,
"ObInitialCompleteMigrationDag: ls_id = %s, migration_type = %s",
to_cstring(self_ctx->arg_.ls_id_), ObMigrationOpType::get_str(self_ctx->arg_.type_)))) {
"ObInitialCompleteMigrationDag: ls_id = %s, migration_type = %s, dag_prio = %s",
to_cstring(self_ctx->arg_.ls_id_), ObMigrationOpType::get_str(self_ctx->arg_.type_),
ObIDag::get_dag_prio_str(this->get_priority())))) {
LOG_WARN("failed to fill comment", K(ret), K(*self_ctx));
}
return ret;
@ -798,6 +802,7 @@ int ObInitialCompleteMigrationTask::generate_migration_dags_()
ObFinishCompleteMigrationDag *finish_complete_dag = nullptr;
ObTenantDagScheduler *scheduler = nullptr;
ObInitialCompleteMigrationDag *initial_complete_migration_dag = nullptr;
ObDagPrio::ObDagPrioEnum prio = ObDagPrio::DAG_PRIO_MAX;
if (!is_inited_) {
ret = OB_NOT_INIT;
@ -808,10 +813,12 @@ int ObInitialCompleteMigrationTask::generate_migration_dags_()
} else if (OB_ISNULL(initial_complete_migration_dag = static_cast<ObInitialCompleteMigrationDag *>(this->get_dag()))) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("initial complete migration dag should not be NULL", K(ret), KP(initial_complete_migration_dag));
} else if (OB_FAIL(ObMigrationUtils::get_dag_priority(ctx_->arg_.type_, prio))) {
LOG_WARN("failed to get dag priority", K(ret));
} else {
if (OB_FAIL(scheduler->alloc_dag(start_complete_dag))) {
if (OB_FAIL(scheduler->alloc_dag_with_priority(prio, start_complete_dag))) {
LOG_WARN("failed to alloc start complete migration dag ", K(ret));
} else if (OB_FAIL(scheduler->alloc_dag(finish_complete_dag))) {
} else if (OB_FAIL(scheduler->alloc_dag_with_priority(prio, finish_complete_dag))) {
LOG_WARN("failed to alloc finish complete migration dag", K(ret));
} else if (OB_FAIL(start_complete_dag->init(dag_net_))) {
LOG_WARN("failed to init start complete migration dag", K(ret));
@ -892,8 +899,9 @@ int ObStartCompleteMigrationDag::fill_dag_key(char *buf, const int64_t buf_len)
LOG_WARN("ha dag net ctx type is unexpected", K(ret), KPC(ha_dag_net_ctx_));
} else if (FALSE_IT(self_ctx = static_cast<ObLSCompleteMigrationCtx *>(ha_dag_net_ctx_))) {
} else if (OB_FAIL(databuff_printf(buf, buf_len,
"ObStartPrepareMigrationDag: ls_id = %s, migration_type = %s",
to_cstring(self_ctx->arg_.ls_id_), ObMigrationOpType::get_str(self_ctx->arg_.type_)))) {
"ObStartPrepareMigrationDag: ls_id = %s, migration_type = %s, dag_prio = %s",
to_cstring(self_ctx->arg_.ls_id_), ObMigrationOpType::get_str(self_ctx->arg_.type_),
ObIDag::get_dag_prio_str(this->get_priority())))) {
LOG_WARN("failed to fill comment", K(ret), K(*self_ctx));
}
return ret;
@ -2094,8 +2102,9 @@ int ObFinishCompleteMigrationDag::fill_dag_key(char *buf, const int64_t buf_len)
LOG_WARN("ha dag net ctx type is unexpected", K(ret), KPC(ha_dag_net_ctx_));
} else if (FALSE_IT(self_ctx = static_cast<ObLSCompleteMigrationCtx *>(ha_dag_net_ctx_))) {
} else if (OB_FAIL(databuff_printf(buf, buf_len,
"ObFinishCompleteMigrationDag: ls_id = %s, migration_type = %s",
to_cstring(self_ctx->arg_.ls_id_), ObMigrationOpType::get_str(self_ctx->arg_.type_)))) {
"ObFinishCompleteMigrationDag: ls_id = %s, migration_type = %s, dag_prio = %s",
to_cstring(self_ctx->arg_.ls_id_), ObMigrationOpType::get_str(self_ctx->arg_.type_),
ObIDag::get_dag_prio_str(this->get_priority())))) {
LOG_WARN("failed to fill comment", K(ret), K(*self_ctx));
}
return ret;
@ -2222,6 +2231,7 @@ int ObFinishCompleteMigrationTask::generate_prepare_initial_dag_()
ObInitialCompleteMigrationDag *initial_complete_dag = nullptr;
ObTenantDagScheduler *scheduler = nullptr;
ObFinishCompleteMigrationDag *finish_complete_migration_dag = nullptr;
ObDagPrio::ObDagPrioEnum prio = ObDagPrio::DAG_PRIO_MAX;
if (!is_inited_) {
ret = OB_NOT_INIT;
@ -2232,8 +2242,10 @@ int ObFinishCompleteMigrationTask::generate_prepare_initial_dag_()
} else if (OB_ISNULL(scheduler = MTL(ObTenantDagScheduler*))) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("failed to get ObTenantDagScheduler from MTL", K(ret));
} else if (OB_FAIL(ObMigrationUtils::get_dag_priority(ctx_->arg_.type_, prio))) {
LOG_WARN("failed to get dag priority", K(ret));
} else {
if (OB_FAIL(scheduler->alloc_dag(initial_complete_dag))) {
if (OB_FAIL(scheduler->alloc_dag_with_priority(prio, initial_complete_dag))) {
LOG_WARN("failed to alloc initial complete migration dag ", K(ret));
} else if (OB_FAIL(initial_complete_dag->init(dag_net_))) {
LOG_WARN("failed to init initial complete migration dag", K(ret));

85
src/storage/high_availability/ob_ls_migration.cpp
View File

@ -311,6 +311,7 @@ int ObMigrationDagNet::start_running_for_migration_()
int tmp_ret = OB_SUCCESS;
ObInitialMigrationDag *initial_migration_dag = nullptr;
ObTenantDagScheduler *scheduler = nullptr;
ObDagPrio::ObDagPrioEnum prio = ObDagPrio::DAG_PRIO_MAX;
if (!is_inited_) {
ret = OB_NOT_INIT;
@ -319,8 +320,10 @@ int ObMigrationDagNet::start_running_for_migration_()
} else if (OB_ISNULL(scheduler = MTL(ObTenantDagScheduler*))) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("failed to get ObTenantDagScheduler from MTL", K(ret));
} else if (OB_FAIL(scheduler->alloc_dag(initial_migration_dag))) {
LOG_WARN("failed to alloc migration dag ", K(ret));
} else if (OB_FAIL(ObMigrationUtils::get_dag_priority(ctx_->arg_.type_, prio))) {
LOG_WARN("failed to get dag priority", K(ret));
} else if (OB_FAIL(scheduler->alloc_dag_with_priority(prio, initial_migration_dag))) {
LOG_WARN("failed to alloc migration dag ", K(ret), K(prio));
} else if (OB_FAIL(initial_migration_dag->init(this))) {
LOG_WARN("failed to init migration dag", K(ret));
} else if (OB_FAIL(add_dag_into_dag_net(*initial_migration_dag))) {
@ -549,8 +552,9 @@ int ObInitialMigrationDag::fill_dag_key(char *buf, const int64_t buf_len) const
ret = OB_ERR_UNEXPECTED;
LOG_WARN("inital migration dag migration ctx should not be NULL", K(ret), KP(ctx));
} else if (OB_FAIL(databuff_printf(buf, buf_len,
"ObInitialMigrationDag: ls_id = %s, migration_type = %s",
to_cstring(ctx->arg_.ls_id_), ObMigrationOpType::get_str(ctx->arg_.type_)))) {
"ObInitialMigrationDag: ls_id = %s, migration_type = %s, dag_prio = %s",
to_cstring(ctx->arg_.ls_id_), ObMigrationOpType::get_str(ctx->arg_.type_),
ObIDag::get_dag_prio_str(this->get_priority())))) {
LOG_WARN("failed to fill comment", K(ret), KPC(ctx));
}
return ret;
@ -703,6 +707,7 @@ int ObInitialMigrationTask::generate_migration_dags_()
ObMigrationFinishDag *migration_finish_dag = nullptr;
ObTenantDagScheduler *scheduler = nullptr;
ObInitialMigrationDag *initial_migration_dag = nullptr;
ObDagPrio::ObDagPrioEnum prio = ObDagPrio::DAG_PRIO_MAX;
if (!is_inited_) {
ret = OB_NOT_INIT;
@ -713,9 +718,11 @@ int ObInitialMigrationTask::generate_migration_dags_()
} else if (OB_ISNULL(initial_migration_dag = static_cast<ObInitialMigrationDag *>(this->get_dag()))) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("initial migration dag is null", K(ret), KP(initial_migration_dag));
} else if (OB_FAIL(scheduler->alloc_dag(start_migration_dag))) {
} else if (OB_FAIL(ObMigrationUtils::get_dag_priority(ctx_->arg_.type_, prio))) {
LOG_WARN("failed to get dag priority", K(ret));
} else if (OB_FAIL(scheduler->alloc_dag_with_priority(prio, start_migration_dag))) {
LOG_WARN("failed to alloc start migration dag ", K(ret));
} else if (OB_FAIL(scheduler->alloc_dag(migration_finish_dag))) {
} else if (OB_FAIL(scheduler->alloc_dag_with_priority(prio, migration_finish_dag))) {
LOG_WARN("failed to alloc migration finish dag", K(ret));
} else if (OB_FAIL(start_migration_dag->init(dag_net_))) {
LOG_WARN("failed to init start migration dag", K(ret));
@ -845,8 +852,9 @@ int ObStartMigrationDag::fill_dag_key(char *buf, const int64_t buf_len) const
ret = OB_ERR_UNEXPECTED;
LOG_WARN("migration ctx should not be NULL", K(ret), KP(ctx));
} else if (OB_FAIL(databuff_printf(buf, buf_len,
"ObStartMigrationDag: ls_id = %s, migration_type = %s",
to_cstring(ctx->arg_.ls_id_), ObMigrationOpType::get_str(ctx->arg_.type_)))) {
"ObStartMigrationDag: ls_id = %s, migration_type = %s, dag_prio = %s",
to_cstring(ctx->arg_.ls_id_), ObMigrationOpType::get_str(ctx->arg_.type_),
ObIDag::get_dag_prio_str(this->get_priority())))) {
LOG_WARN("failed to fill comment", K(ret), KPC(ctx));
}
return ret;
@ -1278,6 +1286,7 @@ int ObStartMigrationTask::generate_tablets_migration_dag_()
ObTenantDagScheduler *scheduler = nullptr;
ObIDagNet *dag_net = nullptr;
ObStartMigrationDag *start_migration_dag = nullptr;
ObDagPrio::ObDagPrioEnum prio = ObDagPrio::DAG_PRIO_MAX;
if (!is_inited_) {
ret = OB_NOT_INIT;
@ -1291,10 +1300,12 @@ int ObStartMigrationTask::generate_tablets_migration_dag_()
} else if (OB_ISNULL(scheduler = MTL(ObTenantDagScheduler*))) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("failed to get ObTenantDagScheduler from MTL", K(ret));
} else if (OB_FAIL(ObMigrationUtils::get_dag_priority(ctx_->arg_.type_, prio))) {
LOG_WARN("failed to get dag priority", K(ret));
} else {
if (OB_FAIL(scheduler->alloc_dag(sys_tablets_migration_dag))) {
if (OB_FAIL(scheduler->alloc_dag_with_priority(prio, sys_tablets_migration_dag))) {
LOG_WARN("failed to alloc sys tablets migration dag ", K(ret));
} else if (OB_FAIL(scheduler->alloc_dag(data_tablets_migration_dag))) {
} else if (OB_FAIL(scheduler->alloc_dag_with_priority(prio, data_tablets_migration_dag))) {
LOG_WARN("failed to alloc data tablets migration dag ", K(ret));
} else if (OB_FAIL(sys_tablets_migration_dag->init(dag_net))) {
LOG_WARN("failed to init sys tablets migration dag", K(ret), K(*ctx_));
@ -1697,8 +1708,9 @@ int ObSysTabletsMigrationDag::fill_dag_key(char *buf, const int64_t buf_len) con
ret = OB_ERR_UNEXPECTED;
LOG_WARN("migration ctx should not be NULL", K(ret), KP(ctx));
} else if (OB_FAIL(databuff_printf(buf, buf_len,
"ObSysTabletsMigrationDag: ls_id = %s, migration_type = %s",
to_cstring(ctx->arg_.ls_id_), ObMigrationOpType::get_str(ctx->arg_.type_)))) {
"ObSysTabletsMigrationDag: ls_id = %s, migration_type = %s, dag_prio = %s",
to_cstring(ctx->arg_.ls_id_), ObMigrationOpType::get_str(ctx->arg_.type_),
ObIDag::get_dag_prio_str(this->get_priority())))) {
LOG_WARN("failed to fill comment", K(ret), KPC(ctx));
}
return ret;
@ -1866,6 +1878,7 @@ int ObSysTabletsMigrationTask::generate_sys_tablet_migartion_dag_()
ObSysTabletsMigrationDag *sys_tablets_migration_dag = nullptr;
ObLS *ls = nullptr;
ObIDag *parent = nullptr;
ObDagPrio::ObDagPrioEnum prio = ObDagPrio::DAG_PRIO_MAX;
if (!is_inited_) {
ret = OB_NOT_INIT;
@ -1885,6 +1898,8 @@ int ObSysTabletsMigrationTask::generate_sys_tablet_migartion_dag_()
} else if (OB_FALSE_IT(parent = this->get_dag())) {
} else if (OB_FAIL(tablet_migration_dag_array.push_back(parent))) {
LOG_WARN("failed to push sys_tablets_migration_dag into array", K(ret), K(*ctx_));
} else if (OB_FAIL(ObMigrationUtils::get_dag_priority(ctx_->arg_.type_, prio))) {
LOG_WARN("failed to get dag priority", K(ret));
} else {
for (int64_t i = 0; OB_SUCC(ret) && i < ctx_->sys_tablet_id_array_.count(); ++i) {
const ObTabletID &tablet_id = ctx_->sys_tablet_id_array_.at(i);
@ -1892,7 +1907,7 @@ int ObSysTabletsMigrationTask::generate_sys_tablet_migartion_dag_()
ObTabletHandle tablet_handle;
if (OB_FAIL(ls->ha_get_tablet(tablet_id, tablet_handle))) {
LOG_WARN("failed to get tablet", K(ret), K(tablet_id));
} else if (OB_FAIL(scheduler->alloc_dag(tablet_migration_dag))) {
} else if (OB_FAIL(scheduler->alloc_dag_with_priority(prio, tablet_migration_dag))) {
LOG_WARN("failed to alloc tablet migration dag ", K(ret));
} else if (OB_FAIL(tablet_migration_dag_array.push_back(tablet_migration_dag))) {
LOG_WARN("failed to push tablet migration dag into array", K(ret), K(*ctx_));
@ -2033,8 +2048,9 @@ int ObTabletMigrationDag::fill_dag_key(char *buf, const int64_t buf_len) const
ret = OB_ERR_UNEXPECTED;
LOG_WARN("migration ctx should not be NULL", K(ret), KP(ctx));
} else if (OB_FAIL(databuff_printf(buf, buf_len,
"ObTabletMigrationDag: log_stream_id = %s, tablet_id = %s, migration_type = %s",
to_cstring(ctx->arg_.ls_id_), to_cstring(copy_tablet_ctx_.tablet_id_), ObMigrationOpType::get_str(ctx->arg_.type_)))) {
"ObTabletMigrationDag: log_stream_id = %s, tablet_id = %s, migration_type = %s, dag_prio = %s",
to_cstring(ctx->arg_.ls_id_), to_cstring(copy_tablet_ctx_.tablet_id_), ObMigrationOpType::get_str(ctx->arg_.type_),
ObIDag::get_dag_prio_str(this->get_priority())))) {
LOG_WARN("failed to fill comment", K(ret), KPC(ctx), K(copy_tablet_ctx_));
}
@ -2156,6 +2172,7 @@ int ObTabletMigrationDag::generate_next_dag(share::ObIDag *&dag)
const int64_t start_ts = ObTimeUtil::current_time();
ObMigrationCtx *ctx = nullptr;
ObLS *ls = nullptr;
ObDagPrio::ObDagPrioEnum prio = ObDagPrio::DAG_PRIO_MAX;
if (!is_inited_) {
ret = OB_NOT_INIT;
@ -2175,6 +2192,8 @@ int ObTabletMigrationDag::generate_next_dag(share::ObIDag *&dag)
} else if (OB_ISNULL(ls = ls_handle_.get_ls())) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("ls should not be NULL", K(ret), KPC(ctx));
} else if (OB_FAIL(ObMigrationUtils::get_dag_priority(ctx->arg_.type_, prio))) {
LOG_WARN("failed to get dag priority", K(ret));
} else {
while (OB_SUCC(ret)) {
ObTabletHandle tablet_handle;
@ -2197,7 +2216,7 @@ int ObTabletMigrationDag::generate_next_dag(share::ObIDag *&dag)
} else {
LOG_WARN("failed to get tablet", K(ret), K(tablet_id));
}
} else if (OB_FAIL(scheduler->alloc_dag(tablet_migration_dag))) {
} else if (OB_FAIL(scheduler->alloc_dag_with_priority(prio, tablet_migration_dag))) {
LOG_WARN("failed to alloc tablet migration dag", K(ret));
} else {
if (OB_FAIL(tablet_migration_dag->init(tablet_id, tablet_handle, dag_net, tablet_group_ctx_))) {
@ -3139,8 +3158,9 @@ int ObDataTabletsMigrationDag::fill_dag_key(char *buf, const int64_t buf_len) co
ret = OB_ERR_UNEXPECTED;
LOG_WARN("migration ctx should not be NULL", K(ret), KP(ctx));
} else if (OB_FAIL(databuff_printf(buf, buf_len,
"ObDataTabletsMigrationDag: log_stream_id = %s, migration_type = %s",
to_cstring(ctx->arg_.ls_id_), ObMigrationOpType::get_str(ctx->arg_.type_)))) {
"ObDataTabletsMigrationDag: log_stream_id = %s, migration_type = %s, dag_prio = %s",
to_cstring(ctx->arg_.ls_id_), ObMigrationOpType::get_str(ctx->arg_.type_),
ObIDag::get_dag_prio_str(this->get_priority())))) {
LOG_WARN("failed to fill comment", K(ret), KPC(ctx));
}
return ret;
@ -3551,6 +3571,7 @@ int ObDataTabletsMigrationTask::generate_tablet_group_dag_()
ObDataTabletsMigrationDag *data_tablets_migration_dag = nullptr;
ObHATabletGroupCtx *tablet_group_ctx = nullptr;
ObArray<ObTabletID> tablet_id_array;
ObDagPrio::ObDagPrioEnum prio = ObDagPrio::DAG_PRIO_MAX;
DEBUG_SYNC(BEFORE_TABLET_GROUP_MIGRATION_GENERATE_NEXT_DAG);
@ -3574,8 +3595,10 @@ int ObDataTabletsMigrationTask::generate_tablet_group_dag_()
} else if (OB_ISNULL(scheduler = MTL(ObTenantDagScheduler*))) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("failed to get ObTenantDagScheduler from MTL", K(ret));
} else if (OB_FAIL(ObMigrationUtils::get_dag_priority(ctx_->arg_.type_, prio))) {
LOG_WARN("failed to get dag priority", K(ret));
} else {
if (OB_FAIL(scheduler->alloc_dag(tablet_group_dag))) {
if (OB_FAIL(scheduler->alloc_dag_with_priority(prio, tablet_group_dag))) {
LOG_WARN("failed to alloc tablet group migration dag ", K(ret));
} else if (OB_FAIL(tablet_group_dag->init(tablet_id_array, dag_net, finish_dag_, tablet_group_ctx))) {
LOG_WARN("failed to init tablet group dag", K(ret), K(tablet_id_array));
@ -3798,9 +3821,9 @@ int ObTabletGroupMigrationDag::fill_dag_key(char *buf, const int64_t buf_len) co
ret = OB_ERR_UNEXPECTED;
LOG_WARN("migration ctx should not be NULL", K(ret), KP(ctx));
} else if (OB_FAIL(databuff_printf(buf, buf_len,
"ObTabletGroupMigrationDag: log_stream_id = %s, migration_type = %s, first_tablet_id = %s",
"ObTabletGroupMigrationDag: log_stream_id = %s, migration_type = %s, first_tablet_id = %s, dag_prio = %s",
to_cstring(ctx->arg_.ls_id_), ObMigrationOpType::get_str(ctx->arg_.type_),
to_cstring(tablet_id_array_.at(0))))) {
to_cstring(tablet_id_array_.at(0)), ObIDag::get_dag_prio_str(this->get_priority())))) {
LOG_WARN("failed to fill comment", K(ret), KPC(ctx));
}
return ret;
@ -3872,6 +3895,7 @@ int ObTabletGroupMigrationDag::generate_next_dag(share::ObIDag *&dag)
ObArray<ObTabletID> tablet_id_array;
ObDagId dag_id;
const int64_t start_ts = ObTimeUtil::current_time();
ObDagPrio::ObDagPrioEnum prio = ObDagPrio::DAG_PRIO_MAX;
if (!is_inited_) {
ret = OB_NOT_INIT;
LOG_WARN("tablet group migration dag do not init", K(ret));
@ -3899,7 +3923,9 @@ int ObTabletGroupMigrationDag::generate_next_dag(share::ObIDag *&dag)
} else if (OB_ISNULL(scheduler = MTL(ObTenantDagScheduler*))) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("failed to get ObTenantDagScheduler from MTL", K(ret));
} else if (OB_FAIL(scheduler->alloc_dag(tablet_group_migration_dag))) {
} else if (OB_FAIL(ObMigrationUtils::get_dag_priority(ctx->arg_.type_, prio))) {
LOG_WARN("failed to get dag priority", K(ret));
} else if (OB_FAIL(scheduler->alloc_dag_with_priority(prio, tablet_group_migration_dag))) {
LOG_WARN("failed to alloc tablet backfill tx migration dag ", K(ret));
} else if (OB_FAIL(tablet_group_migration_dag->init(tablet_id_array, dag_net, finish_dag_, tablet_group_ctx))) {
LOG_WARN("failed to init tablet migration dag", K(ret), KPC(ctx));
@ -4075,6 +4101,7 @@ int ObTabletGroupMigrationTask::generate_tablet_migration_dag_()
ObTabletGroupMigrationDag *tablet_group_migration_dag = nullptr;
ObTabletMigrationDag *tablet_migration_dag = nullptr;
ObLS *ls = nullptr;
ObDagPrio::ObDagPrioEnum prio = ObDagPrio::DAG_PRIO_MAX;
DEBUG_SYNC(BEFORE_TABLET_MIGRATION_GENERATE_NEXT_DAG);
@ -4093,6 +4120,8 @@ int ObTabletGroupMigrationTask::generate_tablet_migration_dag_()
} else if (OB_ISNULL(ls = ls_handle_.get_ls())) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("ls should not be NULL", K(ret), KP(ls));
} else if (OB_FAIL(ObMigrationUtils::get_dag_priority(ctx_->arg_.type_, prio))) {
LOG_WARN("failed to get dag priority", K(ret));
} else {
ObIDag *parent = this->get_dag();
ObTabletID tablet_id;
@ -4112,7 +4141,7 @@ int ObTabletGroupMigrationTask::generate_tablet_migration_dag_()
} else {
LOG_WARN("failed to get tablet", K(ret), K(tablet_id));
}
} else if (OB_FAIL(scheduler->alloc_dag(tablet_migration_dag))) {
} else if (OB_FAIL(scheduler->alloc_dag_with_priority(prio, tablet_migration_dag))) {
LOG_WARN("failed to alloc tablet migration dag ", K(ret));
} else if (OB_FAIL(tablet_migration_dag->init(tablet_id, tablet_handle, dag_net, tablet_group_ctx_))) {
LOG_WARN("failed to init tablet migration migration dag", K(ret), K(*ctx_));
@ -4287,8 +4316,9 @@ int ObMigrationFinishDag::fill_dag_key(char *buf, const int64_t buf_len) const
ret = OB_ERR_UNEXPECTED;
LOG_WARN("migration ctx should not be NULL", K(ret), KP(ctx));
} else if (OB_FAIL(databuff_printf(buf, buf_len,
"ObMigrationFinishDag: ls_id = %s, migration_type = %s",
to_cstring(ctx->arg_.ls_id_), ObMigrationOpType::get_str(ctx->arg_.type_)))) {
"ObMigrationFinishDag: ls_id = %s, migration_type = %s, dag_prio = %s",
to_cstring(ctx->arg_.ls_id_), ObMigrationOpType::get_str(ctx->arg_.type_),
ObIDag::get_dag_prio_str(this->get_priority())))) {
LOG_WARN("failed to fill comment", K(ret), KPC(ctx));
}
return ret;
@ -4412,6 +4442,7 @@ int ObMigrationFinishTask::generate_migration_init_dag_()
ObInitialMigrationDag *initial_migration_dag = nullptr;
ObTenantDagScheduler *scheduler = nullptr;
ObMigrationFinishDag *migration_finish_dag = nullptr;
ObDagPrio::ObDagPrioEnum prio = ObDagPrio::DAG_PRIO_MAX;
if (!is_inited_) {
ret = OB_NOT_INIT;
@ -4422,8 +4453,10 @@ int ObMigrationFinishTask::generate_migration_init_dag_()
} else if (OB_ISNULL(migration_finish_dag = static_cast<ObMigrationFinishDag *>(this->get_dag()))) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("migration finish dag should not be NULL", K(ret), KP(migration_finish_dag));
} else if (OB_FAIL(ObMigrationUtils::get_dag_priority(ctx_->arg_.type_, prio))) {
LOG_WARN("failed to get dag priority", K(ret));
} else {
if (OB_FAIL(scheduler->alloc_dag(initial_migration_dag))) {
if (OB_FAIL(scheduler->alloc_dag_with_priority(prio, initial_migration_dag))) {
LOG_WARN("failed to alloc initial migration dag ", K(ret));
} else if (OB_FAIL(initial_migration_dag->init(dag_net_))) {
LOG_WARN("failed to init initial migration dag", K(ret));

39
src/storage/high_availability/ob_ls_prepare_migration.cpp
View File

@ -156,6 +156,7 @@ int ObLSPrepareMigrationDagNet::start_running_for_migration_()
int tmp_ret = OB_SUCCESS;
ObInitialPrepareMigrationDag *initial_dag = nullptr;
ObTenantDagScheduler *scheduler = nullptr;
ObDagPrio::ObDagPrioEnum prio = ObDagPrio::DAG_PRIO_MAX;
if (!is_inited_) {
ret = OB_NOT_INIT;
@ -164,7 +165,9 @@ int ObLSPrepareMigrationDagNet::start_running_for_migration_()
} else if (OB_ISNULL(scheduler = MTL(ObTenantDagScheduler*))) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("failed to get ObTenantDagScheduler from MTL", K(ret));
} else if (OB_FAIL(scheduler->alloc_dag(initial_dag))) {
} else if (OB_FAIL(ObMigrationUtils::get_dag_priority(ctx_.arg_.type_, prio))) {
LOG_WARN("failed to get dag priority", K(ret));
} else if (OB_FAIL(scheduler->alloc_dag_with_priority(prio, initial_dag))) {
LOG_WARN("failed to alloc initial dag ", K(ret));
} else if (OB_FAIL(initial_dag->init(this))) {
LOG_WARN("failed to init initial dag", K(ret));
@ -425,8 +428,9 @@ int ObInitialPrepareMigrationDag::fill_dag_key(char *buf, const int64_t buf_len)
LOG_WARN("ha dag net ctx type is unexpected", K(ret), KPC(ha_dag_net_ctx_));
} else if (FALSE_IT(self_ctx = static_cast<ObLSPrepareMigrationCtx *>(ha_dag_net_ctx_))) {
} else if (OB_FAIL(databuff_printf(buf, buf_len,
"ObInitialPrepareMigrationDag: ls_id = %s, migration_type = %s",
to_cstring(self_ctx->arg_.ls_id_), ObMigrationOpType::get_str(self_ctx->arg_.type_)))) {
"ObInitialPrepareMigrationDag: ls_id = %s, migration_type = %s, dag_prio = %s",
to_cstring(self_ctx->arg_.ls_id_), ObMigrationOpType::get_str(self_ctx->arg_.type_),
ObIDag::get_dag_prio_str(this->get_priority())))) {
LOG_WARN("failed to fill comment", K(ret), K(*self_ctx));
}
return ret;
@ -537,6 +541,7 @@ int ObInitialPrepareMigrationTask::generate_migration_dags_()
ObFinishPrepareMigrationDag *finish_prepare_dag = nullptr;
ObTenantDagScheduler *scheduler = nullptr;
ObInitialPrepareMigrationDag *initial_prepare_migration_dag = nullptr;
ObDagPrio::ObDagPrioEnum prio = ObDagPrio::DAG_PRIO_MAX;
if (!is_inited_) {
ret = OB_NOT_INIT;
@ -547,10 +552,12 @@ int ObInitialPrepareMigrationTask::generate_migration_dags_()
} else if (OB_ISNULL(initial_prepare_migration_dag = static_cast<ObInitialPrepareMigrationDag *>(this->get_dag()))) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("initial prepare migration dag should not be NULL", K(ret), KP(initial_prepare_migration_dag));
} else if (OB_FAIL(ObMigrationUtils::get_dag_priority(ctx_->arg_.type_, prio))) {
LOG_WARN("failed to get dag priority", K(ret));
} else {
if (OB_FAIL(scheduler->alloc_dag(start_prepare_dag))) {
if (OB_FAIL(scheduler->alloc_dag_with_priority(prio, start_prepare_dag))) {
LOG_WARN("failed to alloc start prepare migration dag ", K(ret));
} else if (OB_FAIL(scheduler->alloc_dag(finish_prepare_dag))) {
} else if (OB_FAIL(scheduler->alloc_dag_with_priority(prio, finish_prepare_dag))) {
LOG_WARN("failed to alloc finish prepare migration dag", K(ret));
} else if (OB_FAIL(start_prepare_dag->init(dag_net_))) {
LOG_WARN("failed to init start prepare migration dag", K(ret));
@ -630,8 +637,9 @@ int ObStartPrepareMigrationDag::fill_dag_key(char *buf, const int64_t buf_len) c
LOG_WARN("ha dag net ctx type is unexpected", K(ret), KPC(ha_dag_net_ctx_));
} else if (FALSE_IT(self_ctx = static_cast<ObLSPrepareMigrationCtx *>(ha_dag_net_ctx_))) {
} else if (OB_FAIL(databuff_printf(buf, buf_len,
"ObStartPrepareMigrationDag: ls_id = %s, migration_type = %s",
to_cstring(self_ctx->arg_.ls_id_), ObMigrationOpType::get_str(self_ctx->arg_.type_)))) {
"ObStartPrepareMigrationDag: ls_id = %s, migration_type = %s, dag_prio = %s",
to_cstring(self_ctx->arg_.ls_id_), ObMigrationOpType::get_str(self_ctx->arg_.type_),
ObIDag::get_dag_prio_str(this->get_priority())))) {
LOG_WARN("failed to fill comment", K(ret), K(*self_ctx));
}
return ret;
@ -973,6 +981,7 @@ int ObStartPrepareMigrationTask::generate_prepare_migration_dags_()
ObLS *ls = nullptr;
common::ObArray<ObTabletBackfillInfo> tablet_infos;
storage::ObTabletBackfillInfo tablet_info;
ObDagPrio::ObDagPrioEnum prio = ObDagPrio::DAG_PRIO_MAX;
if (!is_inited_) {
ret = OB_NOT_INIT;
LOG_WARN("start prepare migration task do not init", K(ret));
@ -992,8 +1001,10 @@ int ObStartPrepareMigrationTask::generate_prepare_migration_dags_()
} else if (OB_ISNULL(scheduler = MTL(ObTenantDagScheduler*))) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("failed to get ObTenantDagScheduler from MTL", K(ret));
} else if (OB_FAIL(ObMigrationUtils::get_dag_priority(ctx_->arg_.type_, prio))) {
LOG_WARN("failed to get dag priority", K(ret));
} else {
if (OB_FAIL(scheduler->alloc_dag(finish_backfill_tx_dag))) {
if (OB_FAIL(scheduler->alloc_dag_with_priority(prio, finish_backfill_tx_dag))) {
LOG_WARN("failed to alloc finish backfill tx migration dag ", K(ret));
} else if (OB_FAIL(build_tablet_backfill_info_(tablet_infos))) {
LOG_WARN("failed to build tablet backfill info", K(ret));
@ -1009,7 +1020,7 @@ int ObStartPrepareMigrationTask::generate_prepare_migration_dags_()
} else {
if (OB_FAIL(backfill_tx_ctx->get_tablet_info(tablet_info))) {
LOG_WARN("failed to get tablet id", K(ret), KPC(ctx_));
} else if (OB_FAIL(scheduler->alloc_dag(tablet_backfill_tx_dag))) {
} else if (OB_FAIL(scheduler->alloc_dag_with_priority(prio, tablet_backfill_tx_dag))) {
LOG_WARN("failed to alloc tablet backfill tx migration dag ", K(ret));
} else if (OB_FAIL(tablet_backfill_tx_dag->init(ctx_->task_id_, ctx_->arg_.ls_id_, tablet_info, ctx_, backfill_tx_ctx))) {
LOG_WARN("failed to init tablet backfill tx dag", K(ret), K(*ctx_));
@ -1408,8 +1419,9 @@ int ObFinishPrepareMigrationDag::fill_dag_key(char *buf, const int64_t buf_len)
LOG_WARN("ha dag net ctx type is unexpected", K(ret), KPC(ha_dag_net_ctx_));
} else if (FALSE_IT(self_ctx = static_cast<ObLSPrepareMigrationCtx *>(ha_dag_net_ctx_))) {
} else if (OB_FAIL(databuff_printf(buf, buf_len,
"ObFinishPrepareMigrationDag: ls_id = %s, migration_type = %s",
to_cstring(self_ctx->arg_.ls_id_), ObMigrationOpType::get_str(self_ctx->arg_.type_)))) {
"ObFinishPrepareMigrationDag: ls_id = %s, migration_type = %s, dag_prio = %s",
to_cstring(self_ctx->arg_.ls_id_), ObMigrationOpType::get_str(self_ctx->arg_.type_),
ObIDag::get_dag_prio_str(this->get_priority())))) {
LOG_WARN("failed to fill comment", K(ret), K(*self_ctx));
}
return ret;
@ -1524,6 +1536,7 @@ int ObFinishPrepareMigrationTask::generate_prepare_initial_dag_()
ObInitialPrepareMigrationDag *initial_prepare_dag = nullptr;
ObTenantDagScheduler *scheduler = nullptr;
ObFinishPrepareMigrationDag *finish_prepare_migration_dag = nullptr;
ObDagPrio::ObDagPrioEnum prio = ObDagPrio::DAG_PRIO_MAX;
if (!is_inited_) {
ret = OB_NOT_INIT;
@ -1534,8 +1547,10 @@ int ObFinishPrepareMigrationTask::generate_prepare_initial_dag_()
} else if (OB_ISNULL(finish_prepare_migration_dag = static_cast<ObFinishPrepareMigrationDag *>(this->get_dag()))) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("finish prepare migration dag should not be NULL", K(ret), KP(finish_prepare_migration_dag));
} else if (OB_FAIL(ObMigrationUtils::get_dag_priority(ctx_->arg_.type_, prio))) {
LOG_WARN("failed to get dag priority", K(ret));
} else {
if (OB_FAIL(scheduler->alloc_dag(initial_prepare_dag))) {
if (OB_FAIL(scheduler->alloc_dag_with_priority(prio, initial_prepare_dag))) {
LOG_WARN("failed to alloc initial prepare migration dag ", K(ret));
} else if (OB_FAIL(initial_prepare_dag->init(dag_net_))) {
LOG_WARN("failed to init initial migration dag", K(ret));

32
src/storage/high_availability/ob_storage_ha_struct.cpp
View File

@ -1109,6 +1109,38 @@ int ObMigrationUtils::get_ls_rebuild_seq(const uint64_t tenant_id,
return ret;
}
int ObMigrationUtils::get_dag_priority(
const ObMigrationOpType::TYPE &type,
ObDagPrio::ObDagPrioEnum &prio)
{
int ret = OB_SUCCESS;
if (!ObMigrationOpType::is_valid(type)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("get invalid argument", K(ret), K(type));
} else {
switch (type) {
case ObMigrationOpType::TYPE::ADD_LS_OP: {
prio = ObDagPrio::DAG_PRIO_HA_HIGH;
break;
}
case ObMigrationOpType::TYPE::MIGRATE_LS_OP: {
prio = ObDagPrio::DAG_PRIO_HA_MID;
break;
}
case ObMigrationOpType::TYPE::REBUILD_LS_OP: {
prio = ObDagPrio::DAG_PRIO_HA_HIGH;
break;
}
default: {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("migration op type not expected", K(ret), K(type));
break;
}
}
}
return ret;
}
/******************ObCopyTableKeyInfo*********************/
ObCopyTableKeyInfo::ObCopyTableKeyInfo()
: src_table_key_(),

4
src/storage/high_availability/ob_storage_ha_struct.h
View File

@ -23,6 +23,7 @@
#include "storage/blocksstable/ob_datum_rowkey.h"
#include "storage/blocksstable/ob_logic_macro_id.h"
#include "share/ls/ob_ls_i_life_manager.h"
#include "share/scheduler/ob_dag_scheduler_config.h"
namespace oceanbase
@ -274,6 +275,9 @@ struct ObMigrationUtils
const uint64_t tenant_id,
const share::ObLSID &ls_id,
int64_t &rebuild_seq);
static int get_dag_priority(
const ObMigrationOpType::TYPE &type,
share::ObDagPrio::ObDagPrioEnum &prio);
};
struct ObCopyTableKeyInfo final