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

Fix direct load rpc get res memory sanity abort

Browse Source
This commit is contained in:
suz-yang
2023-07-13 07:12:11 +00:00
committed by ob-robot
parent f109493977
commit d661d9bb89
2 changed files with 45 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

86
src/observer/table_load/client/ob_table_direct_load_rpc_proxy.h
View File

@ -36,53 +36,49 @@ public:
{
};
#define OB_DEFINE_TABLE_DIRECT_LOAD_RPC_CALL_1(name, pcode, Arg) \
int name(const Arg &arg) \
{ \
int ret = OB_SUCCESS; \
table::ObTableDirectLoadRequest request; \
table::ObTableDirectLoadResult result; \
request.credential_ = credential_; \
request.operation_type_ = pcode; \
if (OB_FAIL(request.set_arg(arg, allocator_))) { \
SERVER_LOG(WARN, "fail to set arg", K(ret), K(arg)); \
} else if (OB_FAIL(rpc_proxy_.to(addr_) \
.timeout(timeout_) \
.by(tenant_id_) \
.direct_load(request, result))) { \
SERVER_LOG(WARN, "fail to rpc call direct load", K(ret), K_(addr), K(request)); \
} else if (OB_UNLIKELY(result.operation_type_ != pcode)) { \
ret = OB_ERR_UNEXPECTED; \
SERVER_LOG(WARN, "unexpected operation type", K(ret), K(request), K(result)); \
} else if (OB_UNLIKELY(!result.res_content_.empty())) { \
ret = OB_ERR_UNEXPECTED; \
SERVER_LOG(WARN, "unexpected non empty res content", K(ret), K(result)); \
} \
return ret; \
#define OB_DEFINE_TABLE_DIRECT_LOAD_RPC_CALL_1(name, pcode, Arg) \
int name(const Arg &arg) \
{ \
int ret = OB_SUCCESS; \
obrpc::ObTableRpcProxy rpc_proxy = rpc_proxy_.to(addr_); \
table::ObTableDirectLoadRequest request; \
table::ObTableDirectLoadResult result; \
request.credential_ = credential_; \
request.operation_type_ = pcode; \
if (OB_FAIL(request.set_arg(arg, allocator_))) { \
SERVER_LOG(WARN, "fail to set arg", K(ret), K(arg)); \
} else if (OB_FAIL(rpc_proxy.timeout(timeout_).by(tenant_id_).direct_load(request, result))) { \
SERVER_LOG(WARN, "fail to rpc call direct load", K(ret), K_(addr), K(request)); \
} else if (OB_UNLIKELY(result.operation_type_ != pcode)) { \
ret = OB_ERR_UNEXPECTED; \
SERVER_LOG(WARN, "unexpected operation type", K(ret), K(request), K(result)); \
} else if (OB_UNLIKELY(!result.res_content_.empty())) { \
ret = OB_ERR_UNEXPECTED; \
SERVER_LOG(WARN, "unexpected non empty res content", K(ret), K(result)); \
} \
return ret; \
}
#define OB_DEFINE_TABLE_DIRECT_LOAD_RPC_CALL_2(name, pcode, Arg, Res) \
int name(const Arg &arg, Res &res) \
{ \
int ret = OB_SUCCESS; \
table::ObTableDirectLoadRequest request; \
table::ObTableDirectLoadResult result; \
request.credential_ = credential_; \
request.operation_type_ = pcode; \
if (OB_FAIL(request.set_arg(arg, allocator_))) { \
SERVER_LOG(WARN, "fail to set arg", K(ret), K(arg)); \
} else if (OB_FAIL(rpc_proxy_.to(addr_) \
.timeout(timeout_) \
.by(tenant_id_) \
.direct_load(request, result))) { \
SERVER_LOG(WARN, "fail to rpc call direct load", K(ret), K_(addr), K(request)); \
} else if (OB_UNLIKELY(result.operation_type_ != pcode)) { \
ret = OB_ERR_UNEXPECTED; \
SERVER_LOG(WARN, "unexpected operation type", K(ret), K(request), K(result)); \
} else if (OB_FAIL(result.get_res(res))) { \
SERVER_LOG(WARN, "fail to get res", K(ret), K(result)); \
} \
return ret; \
#define OB_DEFINE_TABLE_DIRECT_LOAD_RPC_CALL_2(name, pcode, Arg, Res) \
int name(const Arg &arg, Res &res) \
{ \
int ret = OB_SUCCESS; \
obrpc::ObTableRpcProxy rpc_proxy = rpc_proxy_.to(addr_); \
table::ObTableDirectLoadRequest request; \
table::ObTableDirectLoadResult result; \
request.credential_ = credential_; \
request.operation_type_ = pcode; \
if (OB_FAIL(request.set_arg(arg, allocator_))) { \
SERVER_LOG(WARN, "fail to set arg", K(ret), K(arg)); \
} else if (OB_FAIL(rpc_proxy.timeout(timeout_).by(tenant_id_).direct_load(request, result))) { \
SERVER_LOG(WARN, "fail to rpc call direct load", K(ret), K_(addr), K(request)); \
} else if (OB_UNLIKELY(result.operation_type_ != pcode)) { \
ret = OB_ERR_UNEXPECTED; \
SERVER_LOG(WARN, "unexpected operation type", K(ret), K(request), K(result)); \
} else if (OB_FAIL(result.get_res(res))) { \
SERVER_LOG(WARN, "fail to get res", K(ret), K(result)); \
} \
return ret; \
}
#define OB_DEFINE_TABLE_DIRECT_LOAD_RPC_CALL(name, pcode, ...) \

8
src/observer/table_load/control/ob_table_load_control_rpc_proxy.h
View File

@ -49,13 +49,13 @@ public:
int name(const Arg &arg) \
{ \
int ret = OB_SUCCESS; \
obrpc::ObSrvRpcProxy rpc_proxy = rpc_proxy_.to(addr_); \
ObDirectLoadControlRequest request; \
ObDirectLoadControlResult result; \
request.command_type_ = pcode; \
if (OB_FAIL(request.set_arg(arg, allocator_))) { \
SERVER_LOG(WARN, "fail to set arg", K(ret), K(arg)); \
} else if (OB_FAIL(rpc_proxy_.to(addr_) \
.timeout(timeout_) \
} else if (OB_FAIL(rpc_proxy.timeout(timeout_) \
.by(tenant_id_) \
.direct_load_control(request, result))) { \
SERVER_LOG(WARN, "fail to rpc call direct load control", K(ret), K_(addr), K(request)); \
@ -73,13 +73,13 @@ public:
int name(const Arg &arg, Res &res) \
{ \
int ret = OB_SUCCESS; \
obrpc::ObSrvRpcProxy rpc_proxy = rpc_proxy_.to(addr_); \
ObDirectLoadControlRequest request; \
ObDirectLoadControlResult result; \
request.command_type_ = pcode; \
if (OB_FAIL(request.set_arg(arg, allocator_))) { \
SERVER_LOG(WARN, "fail to set arg", K(ret), K(arg)); \
} else if (OB_FAIL(rpc_proxy_.to(addr_) \
.timeout(timeout_) \
} else if (OB_FAIL(rpc_proxy.timeout(timeout_) \
.by(tenant_id_) \
.direct_load_control(request, result))) { \
SERVER_LOG(WARN, "fail to rpc call direct load control", K(ret), K_(addr), K(request)); \