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

fix direct load memory leak when timeout

Browse Source
This commit is contained in:
obdev 2023-05-05 07:38:33 +00:00 committed by ob-robot
parent 51d7205f5e
commit d10b15b188
3 changed files with 83 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

112
src/observer/table_load/ob_table_load_store.cpp
View File

@ -871,6 +871,7 @@ int ObTableLoadStore::px_finish_trans(const ObTableLoadTransId &trans_id)
ret = OB_NOT_INIT;
LOG_WARN("ObTableLoadStore not init", KR(ret), KP(this));
} else {
LOG_INFO("store px finish trans", K(trans_id));
ObTableLoadStoreTrans *trans = nullptr;
if (OB_FAIL(store_ctx_->get_segment_trans(trans_id.segment_id_, trans))) {
LOG_WARN("fail to get segment trans", KR(ret));
@ -892,39 +893,6 @@ int ObTableLoadStore::px_finish_trans(const ObTableLoadTransId &trans_id)
return ret;
}
int ObTableLoadStore::px_abandon_trans(const ObTableLoadTransId &trans_id)
{
int ret = OB_SUCCESS;
if (IS_NOT_INIT) {
ret = OB_NOT_INIT;
LOG_WARN("ObTableLoadStore not init", KR(ret), KP(this));
} else {
LOG_INFO("store px abandon trans", K(trans_id));
ObTableLoadStoreTrans *trans = nullptr;
if (OB_FAIL(store_ctx_->get_segment_trans(trans_id.segment_id_, trans))) {
if (OB_UNLIKELY(OB_ENTRY_NOT_EXIST != ret)) {
LOG_WARN("fail to get segment trans", KR(ret));
} else {
ret = OB_SUCCESS;
}
} else if (OB_UNLIKELY(trans_id != trans->get_trans_id())) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("unexpected trans id", KR(ret), K(trans_id), KPC(trans));
} else if (OB_FAIL(trans->set_trans_status_abort())) {
LOG_WARN("fail to set trans status abort", KR(ret));
} else if (OB_FAIL(store_ctx_->abort_trans(trans))) {
LOG_WARN("fail to abort trans", KR(ret));
} else if (OB_FAIL(px_clean_up_trans(trans))) {
LOG_WARN("fail to clean up trans", KR(ret));
}
if (OB_NOT_NULL(trans)) {
store_ctx_->put_trans(trans);
trans = nullptr;
}
}
return ret;
}
int ObTableLoadStore::px_write(const ObTableLoadTransId &trans_id,
const ObTabletID &tablet_id, const ObIArray<ObNewRow> &row_array)
{
@ -964,27 +932,6 @@ int ObTableLoadStore::px_write(const ObTableLoadTransId &trans_id,
return ret;
}
int ObTableLoadStore::px_clean_up_trans(ObTableLoadStoreTrans *trans)
{
int ret = OB_SUCCESS;
if (IS_NOT_INIT) {
ret = OB_NOT_INIT;
LOG_WARN("ObTableLoadStore not init", KR(ret), KP(this));
} else {
ObTableLoadTransStoreWriter *store_writer = nullptr;
if (OB_FAIL(trans->get_store_writer(store_writer))) {
LOG_WARN("fail to get store writer", KR(ret));
} else if (OB_FAIL(store_writer->clean_up(trans->get_trans_id().segment_id_.id_))) {
LOG_WARN("fail to clean up store writer", KR(ret));
}
if (OB_NOT_NULL(store_writer)) {
trans->put_store_writer(store_writer);
store_writer = nullptr;
}
}
return ret;
}
int ObTableLoadStore::px_flush(ObTableLoadStoreTrans *trans)
{
int ret = OB_SUCCESS;
@ -1012,5 +959,62 @@ int ObTableLoadStore::px_flush(ObTableLoadStoreTrans *trans)
return ret;
}
int ObTableLoadStore::px_abandon_trans(ObTableLoadTableCtx *ctx, const ObTableLoadTransId &trans_id)
{
int ret = OB_SUCCESS;
if (OB_UNLIKELY(!ctx->is_valid())) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid args", KR(ret), KPC(ctx));
} else if (OB_UNLIKELY(nullptr == ctx->store_ctx_ || !ctx->store_ctx_->is_valid())) {
// store ctx not init, do nothing
} else {
LOG_INFO("store px abandon trans", K(trans_id));
ObTableLoadStoreCtx *store_ctx = ctx->store_ctx_;
ObTableLoadStoreTrans *trans = nullptr;
if (OB_FAIL(store_ctx->get_segment_trans(trans_id.segment_id_, trans))) {
if (OB_UNLIKELY(OB_ENTRY_NOT_EXIST != ret)) {
LOG_WARN("fail to get segment trans", KR(ret));
} else {
ret = OB_SUCCESS;
}
} else if (OB_UNLIKELY(trans_id != trans->get_trans_id())) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("unexpected trans id", KR(ret), K(trans_id), KPC(trans));
} else if (OB_FAIL(trans->set_trans_status_abort())) {
LOG_WARN("fail to set trans status abort", KR(ret));
} else if (OB_FAIL(store_ctx->abort_trans(trans))) {
LOG_WARN("fail to abort trans", KR(ret));
} else if (OB_FAIL(px_clean_up_trans(trans))) {
LOG_WARN("fail to clean up trans", KR(ret));
}
if (OB_NOT_NULL(trans)) {
store_ctx->put_trans(trans);
trans = nullptr;
}
}
return ret;
}
int ObTableLoadStore::px_clean_up_trans(ObTableLoadStoreTrans *trans)
{
int ret = OB_SUCCESS;
if (OB_ISNULL(trans)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid args", KR(ret), KP(trans));
} else {
ObTableLoadTransStoreWriter *store_writer = nullptr;
if (OB_FAIL(trans->get_store_writer(store_writer))) {
LOG_WARN("fail to get store writer", KR(ret));
} else if (OB_FAIL(store_writer->clean_up(trans->get_trans_id().segment_id_.id_))) {
LOG_WARN("fail to clean up store writer", KR(ret));
}
if (OB_NOT_NULL(store_writer)) {
trans->put_store_writer(store_writer);
store_writer = nullptr;
}
}
return ret;
}
} // namespace observer
} // namespace oceanbase

4
src/observer/table_load/ob_table_load_store.h
View File

@ -73,13 +73,13 @@ private:
public:
int px_start_trans(const table::ObTableLoadTransId &trans_id);
int px_finish_trans(const table::ObTableLoadTransId &trans_id);
int px_abandon_trans(const table::ObTableLoadTransId &trans_id);
int px_write(const table::ObTableLoadTransId &trans_id,
const ObTabletID &tablet_id,
const common::ObIArray<common::ObNewRow> &row_array);
static int px_abandon_trans(ObTableLoadTableCtx *ctx, const table::ObTableLoadTransId &trans_id);
private:
int px_clean_up_trans(ObTableLoadStoreTrans *trans);
int px_flush(ObTableLoadStoreTrans *trans);
static int px_clean_up_trans(ObTableLoadStoreTrans *trans);
private:
ObTableLoadTableCtx * const ctx_;

47
src/sql/engine/cmd/ob_table_direct_insert_service.cpp
View File

@ -108,37 +108,36 @@ int ObTableDirectInsertService::open_task(const uint64_t table_id, const int64_t
}
int ObTableDirectInsertService::close_task(const uint64_t table_id,
const int64_t task_id,
const int error_code)
const int64_t task_id,
const int error_code)
{
int ret = OB_SUCCESS;
ObTableLoadTableCtx *table_ctx = nullptr;
ObTableLoadKey key(MTL_ID(), table_id);
if (OB_FAIL(ObTableLoadService::get_ctx(key, table_ctx))) {
LOG_WARN("fail to get table ctx", KR(ret), K(key), K(table_id));
} else {
table::ObTableLoadTransId trans_id;
trans_id.segment_id_ = task_id;
trans_id.trans_gid_ = 1;
ObTableLoadStore store(table_ctx);
if (OB_FAIL(store.init())) {
LOG_WARN("fail to init store", KR(ret));
if (OB_SUCC(error_code)) {
ObTableLoadTableCtx *table_ctx = nullptr;
ObTableLoadKey key(MTL_ID(), table_id);
if (OB_FAIL(ObTableLoadService::get_ctx(key, table_ctx))) {
LOG_WARN("fail to get table ctx", KR(ret), K(key), K(table_id));
} else {
if (OB_SUCC(error_code)) {
if (OB_FAIL(store.px_finish_trans(trans_id))) {
LOG_WARN("fail to finish direct load trans", KR(ret), K(trans_id));
}
table::ObTableLoadTransId trans_id;
trans_id.segment_id_ = task_id;
trans_id.trans_gid_ = 1;
ObTableLoadStore store(table_ctx);
if (OB_FAIL(store.init())) {
LOG_WARN("fail to init store", KR(ret));
} else if (OB_FAIL(store.px_finish_trans(trans_id))) {
LOG_WARN("fail to finish direct load trans", KR(ret), K(trans_id));
}
if (OB_FAIL(ret)){
if (OB_FAIL(store.px_abandon_trans(trans_id))) {
LOG_WARN("fail to abandon direct load trans", KR(ret));
if (OB_FAIL(ret)) {
int tmp_ret = OB_SUCCESS;
if (OB_TMP_FAIL(ObTableLoadStore::px_abandon_trans(table_ctx, trans_id))) {
LOG_WARN("fail to abandon direct load trans", KR(tmp_ret), K(trans_id));
}
}
}
}
if (OB_NOT_NULL(table_ctx)) {
ObTableLoadService::put_ctx(table_ctx);
table_ctx = nullptr;
if (OB_NOT_NULL(table_ctx)) {
ObTableLoadService::put_ctx(table_ctx);
table_ctx = nullptr;
}
}
return ret;
}