436 lines
14 KiB
C++
436 lines
14 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.
|
|
*/
|
|
|
|
#define USING_LOG_PREFIX STORAGE
|
|
#include <gtest/gtest.h>
|
|
#define private public
|
|
#define protected public
|
|
|
|
#include <thread>
|
|
#include <vector>
|
|
#include <gtest/gtest.h>
|
|
#include "test_backup.h"
|
|
#include "storage/backup/ob_backup_utils.h"
|
|
#include "storage/blocksstable/ob_logic_macro_id.h"
|
|
#include "storage/ob_parallel_external_sort.h"
|
|
#include "storage/blocksstable/ob_data_file_prepare.h"
|
|
|
|
using namespace oceanbase;
|
|
using namespace oceanbase::common;
|
|
using namespace oceanbase::share;
|
|
using namespace oceanbase::backup;
|
|
using namespace oceanbase::blocksstable;
|
|
|
|
namespace oceanbase {
|
|
namespace backup {
|
|
|
|
static ObSimpleMemLimitGetter getter;
|
|
|
|
class ObFakeBackupTabletProvider : public ObIBackupTabletProvider {
|
|
public:
|
|
ObFakeBackupTabletProvider();
|
|
virtual ~ObFakeBackupTabletProvider();
|
|
int init(const int64_t total_item_count);
|
|
void reset() override;
|
|
void reuse() override;
|
|
bool is_run_out() override;
|
|
void set_backup_data_type(const share::ObBackupDataType &backup_data_type)
|
|
{
|
|
UNUSED(backup_data_type);
|
|
}
|
|
share::ObBackupDataType get_backup_data_type() const override
|
|
{
|
|
ObBackupDataType type;
|
|
return type;
|
|
}
|
|
int get_next_batch_items(common::ObIArray<ObBackupProviderItem> &items, int64_t &task_id) override;
|
|
ObBackupTabletProviderType get_type() const override
|
|
{
|
|
return BACKUP_TABLET_PROVIDER;
|
|
}
|
|
|
|
private:
|
|
bool is_inited_;
|
|
lib::ObMutex mutex_;
|
|
int64_t task_id_;
|
|
int64_t supplied_item_count_;
|
|
int64_t total_item_count_;
|
|
DISALLOW_COPY_AND_ASSIGN(ObFakeBackupTabletProvider);
|
|
};
|
|
|
|
ObFakeBackupTabletProvider::ObFakeBackupTabletProvider()
|
|
: is_inited_(false), mutex_(), task_id_(-1), supplied_item_count_(0), total_item_count_(0)
|
|
{}
|
|
|
|
ObFakeBackupTabletProvider::~ObFakeBackupTabletProvider()
|
|
{}
|
|
|
|
int ObFakeBackupTabletProvider::init(const int64_t total_item_count)
|
|
{
|
|
int ret = OB_SUCCESS;
|
|
if (IS_INIT) {
|
|
ret = OB_INIT_TWICE;
|
|
LOG_WARN("init twice", K(ret));
|
|
} else if (total_item_count <= 0) {
|
|
ret = OB_INVALID_ARGUMENT;
|
|
LOG_WARN("get invalid args", K(ret), K(total_item_count));
|
|
} else {
|
|
task_id_ = 0;
|
|
supplied_item_count_ = 0;
|
|
total_item_count_ = total_item_count;
|
|
is_inited_ = true;
|
|
}
|
|
return ret;
|
|
}
|
|
|
|
void ObFakeBackupTabletProvider::reset()
|
|
{
|
|
is_inited_ = false;
|
|
}
|
|
|
|
void ObFakeBackupTabletProvider::reuse()
|
|
{
|
|
task_id_ = 0;
|
|
supplied_item_count_ = 0;
|
|
}
|
|
|
|
bool ObFakeBackupTabletProvider::is_run_out()
|
|
{
|
|
bool run_out = false;
|
|
lib::ObMutexGuard guard(mutex_);
|
|
run_out = supplied_item_count_ >= total_item_count_;
|
|
return run_out;
|
|
}
|
|
|
|
int ObFakeBackupTabletProvider::get_next_batch_items(common::ObIArray<ObBackupProviderItem> &items, int64_t &task_id)
|
|
{
|
|
int ret = OB_SUCCESS;
|
|
items.reset();
|
|
lib::ObMutexGuard guard(mutex_);
|
|
if (supplied_item_count_ >= total_item_count_) {
|
|
ret = OB_ITER_END;
|
|
} else {
|
|
int64_t i = 0;
|
|
int64_t total = random(256, 1024);
|
|
while (OB_SUCC(ret) && i < total) {
|
|
blocksstable::ObLogicMacroBlockId fake_logic_id;
|
|
blocksstable::MacroBlockId fake_macro_block_id;
|
|
storage::ObITable::TableKey fake_table_key;
|
|
common::ObTabletID tablet_id(supplied_item_count_ + 1);
|
|
ObBackupProviderItem item;
|
|
ObBackupMacroBlockId backup_macro_id;
|
|
backup_macro_id.macro_block_id_ = fake_macro_block_id;
|
|
backup_macro_id.logic_id_ = fake_logic_id;
|
|
if (OB_FAIL(item.set(PROVIDER_ITEM_TABLET_META, backup_macro_id, fake_table_key, tablet_id))) {
|
|
LOG_WARN("failed to set item", K(ret));
|
|
} else if (OB_FAIL(items.push_back(item))) {
|
|
LOG_WARN("failed to push back", K(ret), K(item));
|
|
} else {
|
|
i++;
|
|
supplied_item_count_++;
|
|
}
|
|
}
|
|
}
|
|
if (OB_SUCC(ret)) {
|
|
task_id = task_id_;
|
|
++task_id_;
|
|
}
|
|
return ret;
|
|
}
|
|
|
|
void process(ObIBackupTabletProvider *provider, ObBackupMacroBlockTaskMgr *task_mgr)
|
|
{
|
|
int ret = OB_SUCCESS;
|
|
int64_t task_id = 0;
|
|
ObArray<ObBackupProviderItem> item_list;
|
|
while (OB_SUCC(ret)) {
|
|
const bool is_run_out = provider->is_run_out();
|
|
if (!is_run_out) {
|
|
if (OB_FAIL(provider->get_next_batch_items(item_list, task_id))) {
|
|
if (OB_ITER_END == ret) {
|
|
ret = OB_SUCCESS;
|
|
} else {
|
|
LOG_WARN("failed to get next batch");
|
|
}
|
|
} else if (OB_FAIL(task_mgr->receive(task_id, item_list))) {
|
|
LOG_WARN("failed to receive from task mgr", K(ret));
|
|
}
|
|
}
|
|
if (OB_SUCC(ret)) {
|
|
ObArray<ObBackupProviderItem> items;
|
|
const bool has_remain = task_mgr->has_remain();
|
|
int64_t file_id = 0;
|
|
if (OB_FAIL(task_mgr->deliver(items, file_id))) {
|
|
if (OB_EAGAIN == ret) {
|
|
if (has_remain && !is_run_out) {
|
|
continue;
|
|
} else if (is_run_out) {
|
|
LOG_INFO("run out", K(items));
|
|
break;
|
|
}
|
|
}
|
|
} else {
|
|
const ObTabletID &first = items.at(0).tablet_id_;
|
|
const ObTabletID &last = items.at(items.count() - 1).tablet_id_;
|
|
LOG_INFO("backup items infos", K(first), K(last), K(items.count()));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
void test_task_mgr(const int64_t batch_count, const int64_t total_item_count, const int64_t concurrency)
|
|
{
|
|
int ret = OB_SUCCESS;
|
|
const int64_t thread_count = concurrency;
|
|
share::ObBackupDataType backup_data_type;
|
|
backup_data_type.set_minor_data_backup();
|
|
ObBackupMacroBlockTaskMgr *task_mgr = new ObBackupMacroBlockTaskMgr;
|
|
ObFakeBackupTabletProvider *provider = new ObFakeBackupTabletProvider;
|
|
ret = task_mgr->init(backup_data_type, batch_count);
|
|
EXPECT_EQ(OB_SUCCESS, ret);
|
|
ret = provider->init(total_item_count);
|
|
EXPECT_EQ(OB_SUCCESS, ret);
|
|
std::vector<std::thread> workers;
|
|
for (int64_t i = 0; i < thread_count; ++i) {
|
|
std::thread th(process, provider, task_mgr);
|
|
workers.push_back(std::move(th));
|
|
}
|
|
for (auto &it : workers) {
|
|
it.join();
|
|
}
|
|
delete task_mgr;
|
|
delete provider;
|
|
}
|
|
|
|
TEST(TestBackupUtils, test_task_mgr)
|
|
{
|
|
test_task_mgr(1024, 10240, 1);
|
|
}
|
|
|
|
void make_macro_block_id_array(const int64_t tablet_id, const int64_t logic_version,
|
|
const common::ObIArray<int64_t> &data_seq_list, common::ObIArray<ObBackupMacroBlockIDPair> &id_list)
|
|
{
|
|
int ret = OB_SUCCESS;
|
|
ObBackupMacroBlockIDPair macro_id;
|
|
for (int64_t i = 0; OB_SUCC(ret) && i < data_seq_list.count(); ++i) {
|
|
const int64_t data_seq = data_seq_list.at(i);
|
|
macro_id.logic_id_ = ObLogicMacroBlockId(data_seq, logic_version, tablet_id);
|
|
ret = id_list.push_back(macro_id);
|
|
ASSERT_EQ(OB_SUCCESS, ret);
|
|
}
|
|
}
|
|
|
|
TEST(TestBackupUtils, test_check_macro_block_reuse)
|
|
{
|
|
int ret = OB_SUCCESS;
|
|
ObBackupTabletProvider provider;
|
|
bool need_skip = false;
|
|
const int64_t tablet_id = 200001;
|
|
const int64_t logic_version_1 = 100;
|
|
common::ObArray<int64_t> data_seq_list;
|
|
ret = data_seq_list.push_back(0);
|
|
ASSERT_EQ(OB_SUCCESS, ret);
|
|
ret = data_seq_list.push_back(1);
|
|
ASSERT_EQ(OB_SUCCESS, ret);
|
|
ret = data_seq_list.push_back(2);
|
|
ASSERT_EQ(OB_SUCCESS, ret);
|
|
ret = data_seq_list.push_back(3);
|
|
ASSERT_EQ(OB_SUCCESS, ret);
|
|
ret = data_seq_list.push_back(5);
|
|
ASSERT_EQ(OB_SUCCESS, ret);
|
|
common::ObArray<ObBackupMacroBlockIDPair> id_pair_list_1;
|
|
|
|
make_macro_block_id_array(tablet_id, logic_version_1, data_seq_list, id_pair_list_1);
|
|
|
|
blocksstable::ObLogicMacroBlockId logic_id_1(0/*data_seq*/, 100/*logic_version*/, tablet_id);
|
|
provider.inner_check_macro_block_need_skip_(logic_id_1, id_pair_list_1, need_skip);
|
|
ASSERT_TRUE(need_skip);
|
|
|
|
logic_id_1 = ObLogicMacroBlockId(1/*data_seq*/, 100/*logic_version*/, tablet_id);
|
|
provider.inner_check_macro_block_need_skip_(logic_id_1, id_pair_list_1, need_skip);
|
|
ASSERT_TRUE(need_skip);
|
|
|
|
logic_id_1 = ObLogicMacroBlockId(2/*data_seq*/, 100/*logic_version*/, tablet_id);
|
|
provider.inner_check_macro_block_need_skip_(logic_id_1, id_pair_list_1, need_skip);
|
|
ASSERT_TRUE(need_skip);
|
|
|
|
logic_id_1 = ObLogicMacroBlockId(3/*data_seq*/, 100/*logic_version*/, tablet_id);
|
|
provider.inner_check_macro_block_need_skip_(logic_id_1, id_pair_list_1, need_skip);
|
|
ASSERT_TRUE(need_skip);
|
|
|
|
logic_id_1 = ObLogicMacroBlockId(4/*data_seq*/, 100/*logic_version*/, tablet_id);
|
|
provider.inner_check_macro_block_need_skip_(logic_id_1, id_pair_list_1, need_skip);
|
|
ASSERT_FALSE(need_skip);
|
|
|
|
logic_id_1 = ObLogicMacroBlockId(5/*data_seq*/, 100/*logic_version*/, tablet_id);
|
|
provider.inner_check_macro_block_need_skip_(logic_id_1, id_pair_list_1, need_skip);
|
|
ASSERT_TRUE(need_skip);
|
|
|
|
logic_id_1 = ObLogicMacroBlockId(6/*data_seq*/, 100/*logic_version*/, tablet_id);
|
|
provider.inner_check_macro_block_need_skip_(logic_id_1, id_pair_list_1, need_skip);
|
|
ASSERT_FALSE(need_skip);
|
|
|
|
logic_id_1 = ObLogicMacroBlockId(0/*data_seq*/, 200/*logic_version*/, tablet_id);
|
|
provider.inner_check_macro_block_need_skip_(logic_id_1, id_pair_list_1, need_skip);
|
|
ASSERT_FALSE(need_skip);
|
|
}
|
|
|
|
class TestBackupExternalSort : public blocksstable::TestDataFilePrepare
|
|
{
|
|
public:
|
|
TestBackupExternalSort();
|
|
virtual ~TestBackupExternalSort() {}
|
|
int init_tenant_mgr();
|
|
void destroy_tenant_mgr();
|
|
int64_t calculate_min_item_count();
|
|
int generate_items(const int64_t min_count, common::ObIAllocator &allocator, ObVector<ObBackupProviderItem *> &items);
|
|
virtual void SetUp();
|
|
virtual void TearDown();
|
|
public:
|
|
static const int64_t MACRO_BLOCK_SIZE = 2 * 1024 * 1024;
|
|
static const int64_t MACRO_BLOCK_COUNT = 15 * 1024;
|
|
private:
|
|
common::ObArenaAllocator allocator_;
|
|
};
|
|
|
|
TestBackupExternalSort::TestBackupExternalSort()
|
|
: TestDataFilePrepare(&getter, "TestBackupExternalSort", MACRO_BLOCK_SIZE, MACRO_BLOCK_COUNT),
|
|
allocator_(ObModIds::TEST)
|
|
{
|
|
}
|
|
|
|
void TestBackupExternalSort::SetUp()
|
|
{
|
|
TestDataFilePrepare::SetUp();
|
|
EXPECT_EQ(OB_SUCCESS, init_tenant_mgr());
|
|
EXPECT_EQ(OB_SUCCESS, ObTmpFileManager::get_instance().init());
|
|
static ObTenantBase tenant_ctx(OB_SYS_TENANT_ID);
|
|
ObTenantEnv::set_tenant(&tenant_ctx);
|
|
ObTenantIOManager *io_service = nullptr;
|
|
EXPECT_EQ(OB_SUCCESS, ObTenantIOManager::mtl_init(io_service));
|
|
}
|
|
|
|
void TestBackupExternalSort::TearDown()
|
|
{
|
|
allocator_.reuse();
|
|
ObTmpFileManager::get_instance().destroy();
|
|
TestDataFilePrepare::TearDown();
|
|
destroy_tenant_mgr();
|
|
}
|
|
|
|
int TestBackupExternalSort::init_tenant_mgr()
|
|
{
|
|
int ret = OB_SUCCESS;
|
|
ObAddr self;
|
|
self.set_ip_addr("127.0.0.1", 8086);
|
|
rpc::frame::ObReqTransport req_transport(NULL, NULL);
|
|
const int64_t ulmt = 128LL << 30;
|
|
const int64_t llmt = 128LL << 30;
|
|
ret = getter.add_tenant(OB_SYS_TENANT_ID, ulmt, llmt);
|
|
EXPECT_EQ(OB_SUCCESS, ret);
|
|
ret = getter.add_tenant(OB_SERVER_TENANT_ID, ulmt, llmt);
|
|
EXPECT_EQ(OB_SUCCESS, ret);
|
|
lib::set_memory_limit(128LL << 32);
|
|
return ret;
|
|
}
|
|
|
|
void TestBackupExternalSort::destroy_tenant_mgr()
|
|
{
|
|
}
|
|
|
|
int64_t TestBackupExternalSort::calculate_min_item_count()
|
|
{
|
|
ObBackupProviderItem item;
|
|
const int64_t min_memory_limit = ObExternalSortConstant::MIN_MEMORY_LIMIT;
|
|
const int64_t item_size = sizeof(item) + item.get_deep_copy_size();
|
|
int64_t min_count = min_memory_limit / item_size;
|
|
LOG_INFO("calculate min item count", K(min_memory_limit), K(item_size), K(min_count));
|
|
return min_count;
|
|
}
|
|
|
|
int TestBackupExternalSort::generate_items(const int64_t min_count, common::ObIAllocator &allocator, ObVector<ObBackupProviderItem *> &items)
|
|
{
|
|
int ret = OB_SUCCESS;
|
|
ObBackupProviderItem *item = NULL;
|
|
items.reset();
|
|
if (min_count < 0) {
|
|
ret = OB_INVALID_ARGUMENT;
|
|
COMMON_LOG(WARN, "invalid argument", K(ret), K(min_count));
|
|
} else {
|
|
void *buf = NULL;
|
|
ObTabletID tablet_id;
|
|
for (int64_t i = 0; OB_SUCC(ret) && i < min_count; ++i) {
|
|
ObBackupProviderItemType item_type = PROVIDER_ITEM_TABLET_META;
|
|
make_random_tablet_id(tablet_id);
|
|
if (OB_ISNULL(buf = allocator.alloc(sizeof(ObBackupProviderItem)))) {
|
|
ret = OB_ALLOCATE_MEMORY_FAILED;
|
|
LOG_WARN("failed to allocate memory", K(ret));
|
|
} else if (OB_ISNULL(item = new (buf) ObBackupProviderItem())) {
|
|
ret = OB_ALLOCATE_MEMORY_FAILED;
|
|
LOG_WARN("failed to placement new item", K(ret));
|
|
} else if (OB_FAIL(item->set_with_fake(item_type, tablet_id))) {
|
|
LOG_WARN("failed to set item", K(ret), K(tablet_id));
|
|
} else if (OB_FAIL(items.push_back(item))) {
|
|
LOG_WARN("failed to push back item", K(ret));
|
|
}
|
|
}
|
|
}
|
|
return ret;
|
|
}
|
|
|
|
TEST_F(TestBackupExternalSort, test_sort)
|
|
{
|
|
int ret = OB_SUCCESS;
|
|
int64_t buf_mem_limit = ObExternalSortConstant::MIN_MEMORY_LIMIT;
|
|
int64_t file_buf_size = 2 << 20;
|
|
int64_t expire_timestamp = 0;
|
|
const uint64_t tenant_id = OB_SYS_TENANT_ID;
|
|
typedef storage::ObExternalSort<ObBackupProviderItem, ObBackupProviderItemCompare> ExternalSort;
|
|
ExternalSort external_sort;
|
|
ObBackupProviderItemCompare backup_item_cmp(ret);
|
|
ObBackupDataType backup_data_type;
|
|
backup_data_type.set_major_data_backup();
|
|
backup_item_cmp.set_backup_data_type(backup_data_type);
|
|
|
|
ObArenaAllocator allocator;
|
|
const int64_t min_count = calculate_min_item_count();
|
|
ObVector<ObBackupProviderItem *> total_items;
|
|
|
|
ret = generate_items(min_count, allocator, total_items);
|
|
EXPECT_EQ(OB_SUCCESS, ret);
|
|
|
|
ret = external_sort.init(buf_mem_limit, file_buf_size, expire_timestamp, tenant_id, &backup_item_cmp);
|
|
EXPECT_EQ(OB_SUCCESS, ret);
|
|
|
|
for (int64_t i = 0; OB_SUCC(ret) && i < total_items.size(); ++i) {
|
|
const ObBackupProviderItem *item = total_items.at(i);
|
|
ret = external_sort.add_item(*item);
|
|
EXPECT_EQ(OB_SUCCESS, ret);
|
|
}
|
|
|
|
ret = external_sort.do_sort(true /*final_merge*/);
|
|
EXPECT_EQ(OB_SUCCESS, ret);
|
|
}
|
|
|
|
} // namespace backup
|
|
} // namespace oceanbase
|
|
|
|
int main(int argc, char **argv)
|
|
{
|
|
system("rm -f test_backup_utils.log*");
|
|
ObLogger &logger = ObLogger::get_logger();
|
|
logger.set_file_name("test_backup_utils.log", true);
|
|
logger.set_log_level("info");
|
|
testing::InitGoogleTest(&argc, argv);
|
|
return RUN_ALL_TESTS();
|
|
}
|