263 lines
8.8 KiB
C++
263 lines
8.8 KiB
C++
/**
|
|
* Copyright (c) 2021 OceanBase
|
|
* OceanBase CE is licensed under Mulan PubL v2.
|
|
* You can use this software according to the terms and conditions of the Mulan PubL v2.
|
|
* You may obtain a copy of Mulan PubL v2 at:
|
|
* http://license.coscl.org.cn/MulanPubL-2.0
|
|
* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
|
|
* EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
|
|
* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
|
|
* See the Mulan PubL v2 for more details.
|
|
*/
|
|
|
|
#include <gtest/gtest.h>
|
|
#include "share/ob_errno.h"
|
|
#include "lib/oblog/ob_log.h"
|
|
#include "lib/net/ob_addr.h"
|
|
#include "storage/tx/ob_timestamp_service.h"
|
|
#include "storage/tx/ob_gts_rpc.h"
|
|
#include "storage/tx/ob_gts_define.h"
|
|
|
|
namespace oceanbase
|
|
{
|
|
using namespace common;
|
|
using namespace transaction;
|
|
using namespace obrpc;
|
|
namespace unittest
|
|
{
|
|
|
|
class MyResponseRpc : public ObIGtsResponseRpc
|
|
{
|
|
public:
|
|
MyResponseRpc() {}
|
|
~MyResponseRpc() {}
|
|
public:
|
|
void set_valid_arg(const uint64_t tenant_id,
|
|
const int status,
|
|
const ObAddr &sender,
|
|
const ObAddr self)
|
|
{
|
|
tenant_id_ = tenant_id;
|
|
status_ = status;
|
|
sender_ = sender;
|
|
self_ = self;
|
|
}
|
|
int post(const uint64_t tenant_id, const ObAddr &server, const ObGtsErrResponse &msg)
|
|
{
|
|
int ret = OB_SUCCESS;
|
|
if (!msg.is_valid() ||
|
|
tenant_id != tenant_id_ ||
|
|
msg.get_status() != status_ ||
|
|
msg.get_sender() != sender_ ||
|
|
server != self_) {
|
|
ret = OB_INVALID_ARGUMENT;
|
|
TRANS_LOG(WARN, "invalid argument", K(ret), K(tenant_id), K(server), K(msg), K(*this));
|
|
}
|
|
return ret;
|
|
}
|
|
TO_STRING_KV(K_(tenant_id), K_(status), K_(sender), K_(self));
|
|
private:
|
|
uint64_t tenant_id_;
|
|
int status_;
|
|
ObAddr sender_;
|
|
ObAddr self_;
|
|
};
|
|
|
|
class MyTimestampService : public ObTimestampService
|
|
{
|
|
public:
|
|
MyTimestampService() {}
|
|
~MyTimestampService() {}
|
|
void init(const ObAddr &self)
|
|
{
|
|
self_ = self;
|
|
service_type_ = ServiceType::TimestampService;
|
|
pre_allocated_range_ = TIMESTAMP_PREALLOCATED_RANGE;
|
|
last_id_ = 0;
|
|
limited_id_ = ObTimeUtility::current_time_ns() + TIMESTAMP_PREALLOCATED_RANGE;
|
|
}
|
|
public:
|
|
int handle_request(const ObGtsRequest &request, ObGtsRpcResult &result)
|
|
{
|
|
static int64_t total_cnt = 0;
|
|
static int64_t total_rt = 0;
|
|
static const int64_t STATISTICS_INTERVAL_US = 10000000;
|
|
const MonotonicTs start = MonotonicTs::current_time();
|
|
int ret = OB_SUCCESS;
|
|
|
|
if (OB_UNLIKELY(!request.is_valid())) {
|
|
ret = OB_INVALID_ARGUMENT;
|
|
TRANS_LOG(WARN, "invalid argument", KR(ret), K(request));
|
|
} else {
|
|
TRANS_LOG(DEBUG, "handle gts request", K(request));
|
|
int64_t gts = 0;
|
|
const MonotonicTs srr = request.get_srr();
|
|
const uint64_t tenant_id = request.get_tenant_id();
|
|
const ObAddr &requester = request.get_sender();
|
|
int64_t end_id;
|
|
if (requester == self_) {
|
|
// Go local call to get gts
|
|
TRANS_LOG(DEBUG, "handle local gts request", K(requester));
|
|
ret = handle_local_request_(request, result);
|
|
} else if (OB_FAIL(get_number(1, ObTimeUtility::current_time_ns(), gts, end_id))) {
|
|
TRANS_LOG(WARN, "get timestamp failed", KR(ret));
|
|
int tmp_ret = OB_SUCCESS;
|
|
ObGtsErrResponse response;
|
|
if (OB_SUCCESS != (tmp_ret = result.init(tenant_id, ret, srr, 0, 0))) {
|
|
TRANS_LOG(WARN, "gts result init failed", K(tmp_ret), K(request));
|
|
} else if (OB_SUCCESS != (tmp_ret = response.init(tenant_id, srr, ret, self_))) {
|
|
TRANS_LOG(WARN, "gts err response init failed", K(tmp_ret), K(request));
|
|
} else if (OB_SUCCESS != (tmp_ret = rpc_.post(tenant_id, requester, response))) {
|
|
TRANS_LOG(WARN, "post gts err response failed", K(tmp_ret), K(response));
|
|
} else {
|
|
TRANS_LOG(DEBUG, "post gts err response success", K(response));
|
|
}
|
|
} else {
|
|
if (OB_FAIL(result.init(tenant_id, ret, srr, gts, gts))) {
|
|
TRANS_LOG(WARN, "gts result init failed", KR(ret), K(request));
|
|
}
|
|
}
|
|
}
|
|
return ret;
|
|
}
|
|
int handle_local_request_(const ObGtsRequest &request, obrpc::ObGtsRpcResult &result)
|
|
{
|
|
int ret = OB_SUCCESS;
|
|
int64_t gts = 0;
|
|
const uint64_t tenant_id = request.get_tenant_id();
|
|
const MonotonicTs srr = request.get_srr();
|
|
int64_t end_id;
|
|
if (OB_FAIL(get_number(1, ObTimeUtility::current_time_ns(), gts, end_id))) {
|
|
TRANS_LOG(WARN, "get timestamp failed", KR(ret));
|
|
int tmp_ret = OB_SUCCESS;
|
|
if (OB_SUCCESS != (tmp_ret = result.init(tenant_id, ret, srr, 0, 0))) {
|
|
TRANS_LOG(WARN, "gts result init failed", K(tmp_ret), K(request));
|
|
}
|
|
} else {
|
|
if (OB_FAIL(result.init(tenant_id, ret, srr, gts, gts))) {
|
|
TRANS_LOG(WARN, "local gts result init failed", KR(ret), K(request));
|
|
}
|
|
}
|
|
return ret;
|
|
}
|
|
private:
|
|
int get_number(const int64_t range, const int64_t base_id, int64_t &start_id, int64_t &end_id)
|
|
{
|
|
int ret = OB_SUCCESS;
|
|
if (range != 1) {
|
|
ret = OB_INVALID_ARGUMENT;
|
|
TRANS_LOG(WARN, "invalid argument", K(ret), K(range), K(base_id));
|
|
} else if (base_id < ObTimeUtility::current_time()) {
|
|
ret = OB_INVALID_ARGUMENT;
|
|
TRANS_LOG(WARN, "invalid argument", K(ret), K(range), K(base_id));
|
|
} else {
|
|
const int64_t last_id = ATOMIC_LOAD(&last_id_);
|
|
int64_t tmp_id = 0;
|
|
if (base_id > last_id) {
|
|
if (ATOMIC_BCAS(&last_id_, last_id, base_id + range)) {
|
|
tmp_id = base_id;
|
|
} else {
|
|
tmp_id = ATOMIC_FAA(&last_id_, range);
|
|
}
|
|
} else {
|
|
tmp_id = ATOMIC_FAA(&last_id_, range);
|
|
}
|
|
start_id = tmp_id;
|
|
}
|
|
return ret;
|
|
}
|
|
MyResponseRpc rpc_;
|
|
};
|
|
|
|
class TestObGtsMgr : public ::testing::Test
|
|
{
|
|
public :
|
|
virtual void SetUp() {}
|
|
virtual void TearDown() {}
|
|
};
|
|
|
|
//////////////////////basic function test//////////////////////////////////////////
|
|
|
|
TEST_F(TestObGtsMgr, handle_gts_request_by_leader)
|
|
{
|
|
TRANS_LOG(INFO, "called", "func", test_info_->name());
|
|
const ObAddr client(ObAddr::IPV4, "10.0.0.1", 10000);
|
|
const ObAddr server(ObAddr::IPV4, "10.0.0.1", 20000);
|
|
MyTimestampService ts_service;
|
|
MyResponseRpc response_rpc;
|
|
ts_service.init(server);
|
|
|
|
const uint64_t tenant_id = 1001;
|
|
const MonotonicTs srr = MonotonicTs::current_time();
|
|
const int64_t ts_range = 1;
|
|
response_rpc.set_valid_arg(tenant_id, OB_SUCCESS, server, client);
|
|
ObGtsRequest request;
|
|
ObGtsRpcResult result;
|
|
EXPECT_EQ(OB_SUCCESS, request.init(tenant_id, srr, ts_range, client));
|
|
EXPECT_EQ(OB_SUCCESS, ts_service.handle_request(request, result));
|
|
EXPECT_EQ(tenant_id, result.get_tenant_id());
|
|
EXPECT_EQ(OB_SUCCESS, result.get_status());
|
|
EXPECT_EQ(srr, result.get_srr());
|
|
EXPECT_EQ(ts_range - 1, result.get_gts_end() - result.get_gts_start());
|
|
//EXPECT_TRUE(result.get_gts_start() >= srr);
|
|
}
|
|
|
|
TEST_F(TestObGtsMgr, handle_local_gts_request)
|
|
{
|
|
TRANS_LOG(INFO, "called", "func", test_info_->name());
|
|
const ObAddr server(ObAddr::IPV4, "10.0.0.1", 20000);
|
|
MyTimestampService ts_service;
|
|
MyResponseRpc response_rpc;
|
|
ts_service.init(server);
|
|
|
|
const uint64_t tenant_id = 1001;
|
|
const MonotonicTs srr = MonotonicTs::current_time();
|
|
const int64_t ts_range = 1;
|
|
response_rpc.set_valid_arg(tenant_id, OB_SUCCESS, server, server);
|
|
ObGtsRequest request;
|
|
ObGtsRpcResult result;
|
|
EXPECT_EQ(OB_SUCCESS, request.init(tenant_id, srr, ts_range, server));
|
|
EXPECT_EQ(OB_SUCCESS, ts_service.handle_request(request, result));
|
|
EXPECT_EQ(tenant_id, result.get_tenant_id());
|
|
EXPECT_EQ(OB_SUCCESS, result.get_status());
|
|
EXPECT_EQ(srr, result.get_srr());
|
|
EXPECT_EQ(ts_range - 1, result.get_gts_end() - result.get_gts_start());
|
|
// EXPECT_TRUE(result.get_gts_start() >= srr);
|
|
}
|
|
|
|
TEST_F(TestObGtsMgr, invalid_argument)
|
|
{
|
|
TRANS_LOG(INFO, "called", "func", test_info_->name());
|
|
const ObAddr client(ObAddr::IPV4, "10.0.0.1", 10000);
|
|
const ObAddr server(ObAddr::IPV4, "10.0.0.1", 20000);
|
|
MyTimestampService ts_service;
|
|
MyResponseRpc response_rpc;
|
|
|
|
const uint64_t tenant_id = 1001;
|
|
const MonotonicTs srr = MonotonicTs::current_time();
|
|
const MonotonicTs stc;
|
|
const int64_t ts_range = 1;
|
|
ObGtsRequest request;
|
|
EXPECT_EQ(OB_INVALID_ARGUMENT, request.init(0, srr, ts_range, client));
|
|
EXPECT_EQ(OB_INVALID_ARGUMENT, request.init(tenant_id, stc, ts_range, client));
|
|
EXPECT_EQ(OB_INVALID_ARGUMENT, request.init(tenant_id, srr, 0, client));
|
|
EXPECT_EQ(OB_INVALID_ARGUMENT, request.init(tenant_id, srr, ts_range, ObAddr()));
|
|
}
|
|
|
|
}//end of unittest
|
|
}//end of oceanbase
|
|
|
|
using namespace oceanbase;
|
|
using namespace oceanbase::common;
|
|
|
|
int main(int argc, char **argv)
|
|
{
|
|
int ret = 1;
|
|
ObLogger &logger = ObLogger::get_logger();
|
|
logger.set_file_name("test_ob_gts_mgr.log", true);
|
|
logger.set_log_level(OB_LOG_LEVEL_INFO);
|
|
testing::InitGoogleTest(&argc, argv);
|
|
ret = RUN_ALL_TESTS();
|
|
return ret;
|
|
}
|