hcq/oceanbase
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity
Files
aosc-generic-25025-dev
oceanbase/mittest/shared_storage/test_ss_execute_checkpoint_task.cpp
LiefB 4d4991fea0 move sensitive_test into core-test
2024-11-27 17:16:17 +00:00

839 lines
42 KiB
C++
Raw Permalink Blame History

/**
* 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.
*/
#ifndef USING_LOG_PREFIX
#define USING_LOG_PREFIX STORAGETEST
#endif
#include "gtest/gtest.h"
#define private public
#define protected public
#include "lib/ob_errno.h"
#include "mittest/mtlenv/mock_tenant_module_env.h"
#include "share/allocator/ob_tenant_mutil_allocator_mgr.h"
#include "test_ss_common_util.h"
#include "storage/shared_storage/micro_cache/task/ob_ss_micro_cache_task_runner.h"
#include "storage/shared_storage/micro_cache/ob_ss_micro_meta_manager.h"
#include "storage/shared_storage/micro_cache/ob_ss_physical_block_manager.h"
namespace oceanbase
{
namespace storage
{
using namespace oceanbase::common;
class TestSSExecuteCheckpointTask : public ::testing::Test
{
public:
TestSSExecuteCheckpointTask() : allocator_(), micro_cache_(nullptr), phy_blk_mgr_(nullptr),
micro_meta_mgr_(nullptr), arc_task_(nullptr), micro_ckpt_task_(nullptr),
blk_ckpt_task_(nullptr), persist_task_(nullptr) {}
virtual ~TestSSExecuteCheckpointTask() {}
static void SetUpTestCase();
static void TearDownTestCase();
virtual void SetUp();
virtual void TearDown();
void restart_micro_cache();
int add_batch_micro_block(const int64_t start_idx, const int64_t total_cnt,
ObArray<ObSSMicroBlockMeta *> &micro_meta_arr);
int alloc_batch_phy_block_to_reuse(const int64_t total_cnt, ObArray<ObSSPhyBlockPersistInfo> &phy_block_arr);
void check_micro_meta(const ObArray<ObSSMicroBlockMeta *> &micro_meta_arr);
void check_phy_blk_info(const ObArray<ObSSPhyBlockPersistInfo> &phy_block_arr, const int64_t increment);
void check_super_block(const ObSSMicroCacheSuperBlock &super_block);
public:
ObArenaAllocator allocator_;
ObSSMicroCache *micro_cache_;
ObSSPhysicalBlockManager *phy_blk_mgr_;
ObSSMicroMetaManager *micro_meta_mgr_;
ObSSReleaseCacheTask *arc_task_;
ObSSExecuteMicroCheckpointTask *micro_ckpt_task_;
ObSSExecuteBlkCheckpointTask *blk_ckpt_task_;
ObSSPersistMicroDataTask *persist_task_;
};
void TestSSExecuteCheckpointTask::SetUpTestCase()
{
GCTX.startup_mode_ = observer::ObServerMode::SHARED_STORAGE_MODE;
EXPECT_EQ(OB_SUCCESS, MockTenantModuleEnv::get_instance().init());
}
void TestSSExecuteCheckpointTask::TearDownTestCase()
{
MockTenantModuleEnv::get_instance().destroy();
}
void TestSSExecuteCheckpointTask::SetUp()
{
ObSSMicroCache *micro_cache = MTL(ObSSMicroCache *);
ASSERT_NE(nullptr, micro_cache);
micro_cache->stop();
micro_cache->wait();
micro_cache->destroy();
ASSERT_EQ(OB_SUCCESS, micro_cache->init(MTL_ID(), (1L << 32)));
micro_cache->start();
micro_cache_ = micro_cache;
phy_blk_mgr_ = &micro_cache->phy_blk_mgr_;
ASSERT_NE(nullptr, phy_blk_mgr_);
micro_meta_mgr_ = &micro_cache->micro_meta_mgr_;
ASSERT_NE(nullptr, micro_meta_mgr_);
arc_task_ = &micro_cache->task_runner_.release_cache_task_;
ASSERT_NE(nullptr, arc_task_);
micro_ckpt_task_ = &micro_cache->task_runner_.micro_ckpt_task_;
ASSERT_NE(nullptr, micro_ckpt_task_);
blk_ckpt_task_ = &micro_cache->task_runner_.blk_ckpt_task_;
ASSERT_NE(nullptr, blk_ckpt_task_);
persist_task_ = &micro_cache->task_runner_.persist_task_;
ASSERT_NE(nullptr, persist_task_);
}
void TestSSExecuteCheckpointTask::TearDown()
{
ObSSMicroCache *micro_cache = MTL(ObSSMicroCache*);
micro_cache->stop();
micro_cache->wait();
micro_cache->destroy();
}
void TestSSExecuteCheckpointTask::restart_micro_cache()
{
ObSSMicroCache *micro_cache = MTL(ObSSMicroCache*);
ObTenantFileManager *tnt_file_mgr = MTL(ObTenantFileManager*);
micro_cache->stop();
micro_cache->wait();
micro_cache->destroy();
tnt_file_mgr->is_cache_file_exist_ = true;
ASSERT_EQ(OB_SUCCESS, micro_cache->init(MTL_ID(), (1L << 32)));
micro_cache->start();
micro_ckpt_task_->is_inited_ = false;
blk_ckpt_task_->is_inited_ = false;
arc_task_->is_inited_ = false;
persist_task_->is_inited_ = false;
}
int TestSSExecuteCheckpointTask::add_batch_micro_block(
const int64_t start_idx,
const int64_t write_blk_cnt,
ObArray<ObSSMicroBlockMeta *> &micro_meta_arr)
{
int ret = OB_SUCCESS;
persist_task_->is_inited_ = true;
ob_usleep(100 * 1000);
const int64_t payload_offset =
ObSSPhyBlockCommonHeader::get_serialize_size() + ObSSNormalPhyBlockHeader::get_fixed_serialize_size();
const int32_t micro_index_size = sizeof(ObSSMicroBlockIndex) + SS_SERIALIZE_EXTRA_BUF_LEN;
const int32_t micro_cnt = 20;
const int32_t micro_size = (DEFAULT_BLOCK_SIZE - payload_offset) / micro_cnt - micro_index_size;
char *data_buf = nullptr;
ObArray<ObSSMicroBlockCacheKey> micro_key_arr;
if (OB_UNLIKELY(start_idx < 0 || write_blk_cnt <= 0)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", KR(ret), K(start_idx), K(write_blk_cnt));
} else if (OB_ISNULL(data_buf = static_cast<char *>(allocator_.alloc(micro_size)))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("fail to allocate memory", KR(ret), K(micro_size));
} else {
MEMSET(data_buf, 'a', micro_size);
const int64_t end_idx = start_idx + write_blk_cnt;
for (int64_t i = start_idx; OB_SUCC(ret) && i < end_idx; ++i) {
const MacroBlockId macro_id = TestSSCommonUtil::gen_macro_block_id(i);
for (int64_t j = 0; OB_SUCC(ret) && j < micro_cnt; ++j) {
const int32_t offset = payload_offset + j * micro_size;
const ObSSMicroBlockCacheKey micro_key = TestSSCommonUtil::gen_phy_micro_key(macro_id, offset, micro_size);
if (OB_FAIL(micro_cache_->add_micro_block_cache(
micro_key, data_buf, micro_size, ObSSMicroCacheAccessType::COMMON_IO_TYPE))) {
LOG_WARN("fail to add micro_block", KR(ret), K(micro_key), KP(data_buf), K(micro_size));
} else if (OB_FAIL(micro_key_arr.push_back(micro_key))) {
LOG_WARN("fail to push micro_key", KR(ret), K(micro_key));
}
}
if (FAILEDx(TestSSCommonUtil::wait_for_persist_task())) {
LOG_WARN("fail to wait for persist task", K(ret));
}
}
persist_task_->is_inited_ = false;
}
for (int64_t i = 0; OB_SUCC(ret) && i < micro_key_arr.count(); ++i) {
ObSSMicroBlockMetaHandle micro_handle;
ObSSMicroBlockMeta *tmp_micro_meta = nullptr;
const ObSSMicroBlockCacheKey &micro_key = micro_key_arr[i];
if (OB_FAIL(micro_meta_mgr_->get_micro_block_meta_handle(micro_key, micro_handle, false))) {
LOG_WARN("fail to get micro_handle", KR(ret), K(micro_key));
} else if (OB_UNLIKELY(!micro_handle.is_valid())) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("micro_handle is invalid", KR(ret), K(micro_handle), KPC(micro_handle.get_ptr()));
} else if (OB_ISNULL(tmp_micro_meta = static_cast<ObSSMicroBlockMeta *>(allocator_.alloc(sizeof(ObSSMicroBlockMeta))))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("fail to allocate memory", KR(ret), K(sizeof(ObSSMicroBlockMeta)));
} else if (OB_FALSE_IT(tmp_micro_meta->reset())) {
} else if (OB_FALSE_IT(*tmp_micro_meta = *micro_handle.get_ptr())) {
} else if (OB_FAIL(micro_meta_arr.push_back(tmp_micro_meta))) {
LOG_WARN("fail to push micro_meta", KR(ret), KPC(tmp_micro_meta));
}
}
return ret;
}
int TestSSExecuteCheckpointTask::alloc_batch_phy_block_to_reuse(
const int64_t total_cnt,
ObArray<ObSSPhyBlockPersistInfo> &phy_block_arr)
{
int ret = OB_SUCCESS;
if (OB_UNLIKELY(total_cnt <= 0)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", KR(ret), K(total_cnt));
} else {
for (int64_t i = 0; OB_SUCC(ret) && i < total_cnt; i++) {
int64_t block_idx = 0;
ObSSPhysicalBlockHandle phy_blk_handle;
if (OB_FAIL(phy_blk_mgr_->alloc_block(block_idx, phy_blk_handle, ObSSPhyBlockType::SS_NORMAL_BLK))) {
LOG_WARN("fail to alloc block", KR(ret), K(block_idx), K(phy_blk_handle));
} else if (OB_UNLIKELY(!phy_blk_handle.is_valid())) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("phy_blk_handle is invalid", KR(ret), K(phy_blk_handle));
} else {
phy_blk_handle.get_ptr()->alloc_time_us_ -= PHY_BLK_MAX_REUSE_TIME; // mock block is reusable
ObSSPhyBlockPersistInfo phy_info(block_idx, phy_blk_handle.get_ptr()->get_reuse_version());
if (OB_FAIL(phy_block_arr.push_back(phy_info))) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("fail to push phy_info", KR(ret), K(phy_info));
}
}
}
}
return ret;
}
void TestSSExecuteCheckpointTask::check_micro_meta(const ObArray<ObSSMicroBlockMeta *> &micro_meta_arr)
{
int ret = OB_SUCCESS;
const int64_t micro_cnt = micro_meta_arr.count();
for (int64_t i = 0; i < micro_cnt; ++i) {
ObSSMicroBlockMeta *old_micro_meta = micro_meta_arr[i];
const ObSSMicroBlockCacheKey &micro_key = old_micro_meta->get_micro_key();
ObSSMicroBlockMetaHandle micro_handle;
ret = micro_meta_mgr_->get_micro_block_meta_handle(micro_key, micro_handle, false);
// ckpt will record all persisted micro_block's meta
if (old_micro_meta->is_persisted()) {
if (OB_FAIL(ret)) {
LOG_WARN("unexpected", KR(ret), K(micro_key), K(i), K(micro_cnt), KPC(old_micro_meta));
}
ASSERT_EQ(OB_SUCCESS, ret);
ASSERT_EQ(true, micro_handle.is_valid());
ObSSMicroBlockMeta *micro_meta = micro_handle.get_ptr();
ASSERT_EQ(micro_meta->reuse_version_, old_micro_meta->reuse_version_);
ASSERT_EQ(micro_meta->data_dest_, old_micro_meta->data_dest_);
ASSERT_EQ(micro_meta->length_, old_micro_meta->length_);
ASSERT_EQ(micro_meta->is_in_l1_, old_micro_meta->is_in_l1_);
ASSERT_EQ(micro_meta->is_in_ghost_, old_micro_meta->is_in_ghost_);
ASSERT_EQ(micro_meta->is_persisted_, old_micro_meta->is_persisted_);
ASSERT_EQ(micro_meta->is_reorganizing_, old_micro_meta->is_reorganizing_);
} else {
ASSERT_EQ(OB_ENTRY_NOT_EXIST, ret);
}
}
}
void TestSSExecuteCheckpointTask::check_phy_blk_info(
const ObArray<ObSSPhyBlockPersistInfo> &phy_block_arr,
const int64_t increment)
{
for (int64_t i = 0; i < phy_block_arr.count(); ++i) {
ObSSPhysicalBlockHandle phy_blk_handle;
const int64_t block_idx = phy_block_arr[i].blk_idx_;
const int64_t old_reuse_version = phy_block_arr[i].reuse_version_;
ASSERT_EQ(OB_SUCCESS, phy_blk_mgr_->get_block_handle(block_idx, phy_blk_handle));
ASSERT_EQ(true, phy_blk_handle.is_valid());
const int64_t reuse_version = phy_blk_handle.get_ptr()->get_reuse_version();
ASSERT_EQ(old_reuse_version + increment, reuse_version);
}
}
void TestSSExecuteCheckpointTask::check_super_block(const ObSSMicroCacheSuperBlock &super_block)
{
const ObSSMicroCacheSuperBlock &cur_super_block = phy_blk_mgr_->super_block_;
ASSERT_EQ(cur_super_block.micro_ckpt_time_us_, super_block.micro_ckpt_time_us_);
ASSERT_EQ(cur_super_block.cache_file_size_, super_block.cache_file_size_);
ASSERT_EQ(cur_super_block.modify_time_us_, super_block.modify_time_us_);
ASSERT_EQ(cur_super_block.micro_ckpt_entry_list_.count(), super_block.micro_ckpt_entry_list_.count());
for (int64_t i = 0; i < super_block.micro_ckpt_entry_list_.count(); ++i) {
ASSERT_EQ(cur_super_block.micro_ckpt_entry_list_[i], super_block.micro_ckpt_entry_list_[i]);
}
ASSERT_EQ(cur_super_block.blk_ckpt_entry_list_.count(), super_block.blk_ckpt_entry_list_.count());
for (int64_t i = 0; i < super_block.blk_ckpt_entry_list_.count(); ++i) {
ASSERT_EQ(cur_super_block.blk_ckpt_entry_list_[i], super_block.blk_ckpt_entry_list_[i]);
}
}
/* This case tests the basic logic of the execute_checkpoint task. */
TEST_F(TestSSExecuteCheckpointTask, test_execute_checkpoint_task)
{
int ret = OB_SUCCESS;
ObSSMicroCache *micro_cache = MTL(ObSSMicroCache *);
ASSERT_NE(nullptr, micro_cache);
const int64_t block_size = micro_cache->phy_block_size_;
ObSSMicroCacheStat &cache_stat = micro_cache->cache_stat_;
ObSSPersistMicroDataTask &persist_task = micro_cache->task_runner_.persist_task_;
ObSSExecuteMicroCheckpointTask &micro_ckpt_task = micro_cache->task_runner_.micro_ckpt_task_;
micro_ckpt_task.is_inited_ = false;
ObSSExecuteBlkCheckpointTask &blk_ckpt_task = micro_cache->task_runner_.blk_ckpt_task_;
blk_ckpt_task.is_inited_ = false;
ObSSReleaseCacheTask &arc_task = micro_cache->task_runner_.release_cache_task_;
arc_task.is_inited_ = false;
ObSSMemDataManager *mem_data_mgr = &(micro_cache->mem_data_mgr_);
ASSERT_NE(nullptr, mem_data_mgr);
ObSSMicroMetaManager *micro_meta_mgr = &(micro_cache->micro_meta_mgr_);
ASSERT_NE(nullptr, micro_meta_mgr);
ObSSARCInfo &arc_info = micro_meta_mgr->arc_info_;
ObSSPhysicalBlockManager *phy_blk_mgr = &(micro_cache->phy_blk_mgr_);
ASSERT_NE(nullptr, phy_blk_mgr);
const int64_t ori_arc_limit = micro_meta_mgr->get_arc_info().limit_;
// 1. execute phy_block checkpoint
micro_ckpt_task.is_inited_ = true;
blk_ckpt_task.is_inited_ = true;
micro_ckpt_task.interval_us_ = 3600 * 1000 * 1000L;
blk_ckpt_task.interval_us_ = 3600 * 1000 * 1000L;
ob_usleep(1000 * 1000);
blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.need_ckpt_ = true;
ASSERT_EQ(false, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.prev_super_block_.is_valid());
ASSERT_EQ(false, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.cur_super_block_.is_valid());
ASSERT_EQ(0, phy_blk_mgr->blk_cnt_info_.phy_ckpt_blk_.used_cnt_);
ASSERT_EQ(0, phy_blk_mgr->blk_cnt_info_.micro_ckpt_blk_.used_cnt_);
ASSERT_EQ(0, phy_blk_mgr->reusable_set_.size());
ASSERT_EQ(OB_SUCCESS, blk_ckpt_task.ckpt_op_.gen_checkpoint());
ASSERT_EQ(phy_blk_mgr->blk_cnt_info_.total_blk_cnt_ - 2, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.ckpt_item_cnt_);
ASSERT_EQ(true, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.prev_super_block_.is_valid());
ASSERT_EQ(0, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.cur_super_block_.micro_ckpt_entry_list_.count());
int64_t blk_ckpt_list_cnt = blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.cur_super_block_.blk_ckpt_entry_list_.count();
ASSERT_LT(0, blk_ckpt_list_cnt);
ASSERT_EQ(phy_blk_mgr->blk_cnt_info_.phy_ckpt_blk_.used_cnt_, blk_ckpt_list_cnt);
ObSEArray<int64_t, 8> blk_ckpt_entry_list;
ASSERT_EQ(OB_SUCCESS, blk_ckpt_entry_list.assign(blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.cur_super_block_.blk_ckpt_entry_list_));
// 2. execute phy_block checkpoint
blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.reuse();
blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.need_ckpt_ = true;
ASSERT_EQ(false, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.prev_super_block_.is_valid());
ASSERT_EQ(false, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.cur_super_block_.is_valid());
ASSERT_EQ(0, phy_blk_mgr->reusable_set_.size());
ASSERT_EQ(OB_SUCCESS, blk_ckpt_task.ckpt_op_.gen_checkpoint());
ASSERT_EQ(phy_blk_mgr->blk_cnt_info_.total_blk_cnt_ - 2, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.ckpt_item_cnt_);
ASSERT_EQ(true, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.cur_super_block_.is_valid());
ASSERT_EQ(true, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.prev_super_block_.is_valid());
ASSERT_EQ(0, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.cur_super_block_.micro_ckpt_entry_list_.count());
ASSERT_EQ(blk_ckpt_list_cnt, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.cur_super_block_.blk_ckpt_entry_list_.count());
ASSERT_EQ(phy_blk_mgr->blk_cnt_info_.phy_ckpt_blk_.used_cnt_, blk_ckpt_list_cnt);
for (int64_t i = 0; i < blk_ckpt_entry_list.count(); ++i) {
const int64_t blk_idx = blk_ckpt_entry_list.at(i);
ObSSPhysicalBlock *phy_blk = phy_blk_mgr->get_phy_block_by_idx_nolock(blk_idx);
ASSERT_NE(nullptr, phy_blk);
ASSERT_EQ(true, phy_blk->is_free_);
ASSERT_EQ(2, phy_blk->reuse_version_);
ASSERT_EQ(0, phy_blk->valid_len_);
}
ASSERT_NE(blk_ckpt_entry_list.at(0), blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.cur_super_block_.blk_ckpt_entry_list_.at(0));
blk_ckpt_entry_list.reset();
ASSERT_EQ(OB_SUCCESS, blk_ckpt_entry_list.assign(blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.cur_super_block_.blk_ckpt_entry_list_));
ASSERT_EQ(blk_ckpt_list_cnt, blk_ckpt_entry_list.count());
// 3. make some micro_data
const int64_t available_block_cnt = phy_blk_mgr->get_free_normal_block_cnt();
const int64_t WRITE_BLK_CNT = 50;
ASSERT_LT(WRITE_BLK_CNT, available_block_cnt);
const int64_t payload_offset = ObSSPhyBlockCommonHeader::get_serialize_size() +
ObSSNormalPhyBlockHeader::get_fixed_serialize_size();
const int32_t micro_index_size = sizeof(ObSSMicroBlockIndex) + SS_SERIALIZE_EXTRA_BUF_LEN;
const int32_t micro_cnt = 20;
const int32_t micro_size = (block_size - payload_offset) / micro_cnt - micro_index_size;
ObArenaAllocator allocator;
char *data_buf = static_cast<char*>(allocator.alloc(micro_size));
ASSERT_NE(nullptr, data_buf);
MEMSET(data_buf, 'a', micro_size);
// 3.1. write 50 fulfilled phy_block
for (int64_t i = 0; i < WRITE_BLK_CNT; ++i) {
MacroBlockId macro_id = TestSSCommonUtil::gen_macro_block_id(i + 1);
for (int32_t j = 0; j < micro_cnt; ++j) {
const int32_t offset = payload_offset + j * micro_size;
ObSSMicroBlockCacheKey micro_key = TestSSCommonUtil::gen_phy_micro_key(macro_id, offset, micro_size);
micro_cache->add_micro_block_cache(micro_key, data_buf, micro_size,
ObSSMicroCacheAccessType::COMMON_IO_TYPE);
}
ASSERT_NE(nullptr, mem_data_mgr->fg_mem_block_);
ASSERT_EQ(true, mem_data_mgr->fg_mem_block_->is_valid()) << i;
ASSERT_EQ(micro_cnt, mem_data_mgr->fg_mem_block_->micro_count_) << i;
ASSERT_EQ(micro_cnt, mem_data_mgr->fg_mem_block_->micro_count_);
ASSERT_EQ(micro_size * micro_cnt, mem_data_mgr->fg_mem_block_->data_size_);
ASSERT_EQ(micro_size * micro_cnt, mem_data_mgr->fg_mem_block_->valid_val_);
ASSERT_EQ(OB_SUCCESS, TestSSCommonUtil::wait_for_persist_task());
}
{
// to sealed the last mem_block
MacroBlockId macro_id = TestSSCommonUtil::gen_macro_block_id(WRITE_BLK_CNT + 1);
const int32_t offset = payload_offset;
ObSSMicroBlockCacheKey micro_key = TestSSCommonUtil::gen_phy_micro_key(macro_id, offset, micro_size);
micro_cache->add_micro_block_cache(micro_key, data_buf, micro_size,
ObSSMicroCacheAccessType::COMMON_IO_TYPE);
}
// record written phy_block count
const int64_t max_retry_cnt = 10;
bool result_match = false;
for (int64_t i = 0; !result_match && i < max_retry_cnt; ++i) {
result_match = (phy_blk_mgr->blk_cnt_info_.normal_blk_.used_cnt_ == WRITE_BLK_CNT);
if (!result_match) {
ASSERT_EQ(OB_SUCCESS, TestSSCommonUtil::wait_for_persist_task());
}
}
ASSERT_EQ(true, result_match);
usleep(1000 * 1000);
// 3.2. evict all micro_block of the first macro_block
int64_t evict_blk_idx = -1;
for (int64_t i = 0; i < 1; ++i) {
MacroBlockId macro_id = TestSSCommonUtil::gen_macro_block_id(i + 1);
for (int32_t j = 0; j < micro_cnt; ++j) {
const int32_t offset = payload_offset + j * micro_size;
ObSSMicroBlockCacheKey micro_key = TestSSCommonUtil::gen_phy_micro_key(macro_id, offset, micro_size);
ObSSMicroBlockMetaHandle micro_meta_handle;
ASSERT_EQ(OB_SUCCESS, micro_meta_mgr->micro_meta_map_.get(&micro_key, micro_meta_handle));
ObSSMicroBlockMeta *micro_meta = micro_meta_handle.get_ptr();
ASSERT_NE(nullptr, micro_meta);
ASSERT_EQ(true, micro_meta->is_in_l1_);
ASSERT_EQ(false, micro_meta->is_in_ghost_);
ASSERT_EQ(true, micro_meta->is_persisted_);
ObSSMicroBlockMeta *tmp_micro_meta = nullptr;
ASSERT_EQ(OB_SUCCESS, TestSSCommonUtil::alloc_micro_block_meta(tmp_micro_meta));
ASSERT_NE(nullptr, tmp_micro_meta);
*tmp_micro_meta = *micro_meta;
ObSSMicroBlockMetaHandle tmp_micro_handle;
tmp_micro_handle.set_ptr(tmp_micro_meta);
ASSERT_EQ(OB_SUCCESS, micro_meta_mgr->try_evict_micro_block_meta(tmp_micro_handle));
ASSERT_EQ(true, micro_meta->is_in_ghost_);
ASSERT_EQ(false, micro_meta->is_valid_field());
int64_t phy_blk_idx = -1;
bool is_empty_blk = false;
ASSERT_EQ(OB_SUCCESS, phy_blk_mgr->update_block_valid_length(tmp_micro_meta->data_dest_, tmp_micro_meta->reuse_version_,
tmp_micro_meta->length_ * -1, phy_blk_idx, is_empty_blk));
evict_blk_idx = tmp_micro_meta->data_dest_ / phy_blk_mgr->block_size_;
tmp_micro_handle.reset();
}
}
ASSERT_NE(-1, evict_blk_idx);
ASSERT_EQ(((WRITE_BLK_CNT - 1) * micro_cnt + 1) * micro_size, arc_info.get_valid_size());
// 3.3. invalidate all micro_block of the second macro_block
int64_t invalid_blk_idx = -1;
for (int64_t i = 1; i < 2; ++i) {
MacroBlockId macro_id = TestSSCommonUtil::gen_macro_block_id(i + 1);
for (int32_t j = 0; j < micro_cnt; ++j) {
const int32_t offset = payload_offset + j * micro_size;
ObSSMicroBlockCacheKey micro_key = TestSSCommonUtil::gen_phy_micro_key(macro_id, offset, micro_size);
ObSSMicroBlockMetaHandle micro_meta_handle;
ASSERT_EQ(OB_SUCCESS, micro_meta_mgr->get_micro_block_meta_handle(micro_key, micro_meta_handle, false));
ASSERT_EQ(true, micro_meta_handle.is_valid());
ObSSMicroBlockMeta *micro_meta = micro_meta_handle.get_ptr();
int64_t phy_blk_idx = -1;
bool is_empty_blk = false;
ASSERT_EQ(OB_SUCCESS, phy_blk_mgr->update_block_valid_length(micro_meta->data_dest_, micro_meta->reuse_version_,
micro_meta->length_ * -1, phy_blk_idx, is_empty_blk));
invalid_blk_idx = micro_meta->data_dest_ / phy_blk_mgr->block_size_;
micro_meta->mark_invalid();
}
}
ASSERT_NE(-1, invalid_blk_idx);
// 3.4. check micro_meta and macro_meta count
ASSERT_EQ(WRITE_BLK_CNT * micro_cnt + 1, micro_meta_mgr->micro_meta_map_.count());
// 4. execute micro_meta checkpoint
// 4.1. check ckpt state
micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.exe_round_ = ObSSExecuteMicroCheckpointOp::MICRO_META_CKPT_INTERVAL_ROUND;
micro_ckpt_task.is_inited_ = true;
ASSERT_EQ(OB_SUCCESS, micro_ckpt_task.ckpt_op_.check_state());
ASSERT_EQ(OB_SUCCESS, phy_blk_mgr->get_ss_super_block(micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.prev_super_block_));
ASSERT_EQ(true, micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.need_ckpt_);
ASSERT_EQ(0, micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.cur_super_block_.micro_ckpt_entry_list_.count());
ASSERT_EQ(0, micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.cur_super_block_.blk_ckpt_entry_list_.count());
ASSERT_EQ(blk_ckpt_list_cnt, micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.prev_super_block_.blk_ckpt_entry_list_.count());
ASSERT_EQ(0, micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.prev_super_block_.micro_ckpt_entry_list_.count());
// 4.2. scan reusable phy_blocks
ASSERT_EQ(2, phy_blk_mgr->reusable_set_.size());
ASSERT_EQ(OB_SUCCESS, phy_blk_mgr->scan_blocks_to_reuse());
ASSERT_EQ(2, phy_blk_mgr->reusable_set_.size());
// 4.3. first, mock all internal phy_block were used, will fail to gen micro_meta ckpt
const int64_t ori_internal_cnt = phy_blk_mgr->blk_cnt_info_.micro_ckpt_blk_.used_cnt_;
phy_blk_mgr->blk_cnt_info_.micro_ckpt_blk_.used_cnt_ = phy_blk_mgr->blk_cnt_info_.micro_ckpt_blk_.total_cnt_;
ASSERT_EQ(OB_SUCCESS, micro_ckpt_task.ckpt_op_.gen_micro_meta_checkpoint());
ASSERT_EQ(true, micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.lack_phy_blk_);
micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.lack_phy_blk_ = false;
micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.ckpt_item_cnt_ = 0;
micro_ckpt_task.ckpt_op_.tablet_cache_info_map_.clear();
ASSERT_EQ(((WRITE_BLK_CNT - 2) * micro_cnt + 1) * micro_size, arc_info.get_valid_size());
phy_blk_mgr->blk_cnt_info_.micro_ckpt_blk_.used_cnt_ = ori_internal_cnt;
// 4.4. normal situation
ObSEArray<uint64_t, 128> reuse_version_arr;
ASSERT_EQ(OB_SUCCESS, phy_blk_mgr->get_block_reuse_version(reuse_version_arr));
ASSERT_EQ(OB_SUCCESS, micro_ckpt_task.ckpt_op_.gen_micro_meta_checkpoint());
ASSERT_EQ(false, micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.lack_phy_blk_);
ASSERT_LT(0, micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.cur_super_block_.micro_ckpt_entry_list_.count());
ASSERT_LT(0, micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.cur_super_block_.micro_ckpt_time_us_);
ASSERT_EQ((WRITE_BLK_CNT - 1) * micro_cnt, micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.ckpt_item_cnt_);
ASSERT_LT(0, micro_ckpt_task.ckpt_op_.tablet_cache_info_map_.size());
ObSSTabletCacheMap::const_iterator iter = micro_ckpt_task.ckpt_op_.tablet_cache_info_map_.begin();
int64_t total_micro_size = 0;
int64_t tablet_cnt = 0;
for (; iter != micro_ckpt_task.ckpt_op_.tablet_cache_info_map_.end(); ++iter) {
total_micro_size += iter->second.get_valid_size();
++tablet_cnt;
}
ASSERT_EQ(arc_info.get_valid_size(), total_micro_size);
// 4.5. update ss_super_block
ASSERT_EQ(OB_SUCCESS, micro_ckpt_task.ckpt_op_.update_super_block(true));
ObSSMicroCacheSuperBlock cur_super_blk;
ASSERT_EQ(OB_SUCCESS, phy_blk_mgr->get_ss_super_block(cur_super_blk));
ASSERT_LT(0, cur_super_blk.micro_ckpt_entry_list_.count());
ASSERT_EQ(blk_ckpt_entry_list.at(0), cur_super_blk.blk_ckpt_entry_list_.at(0));
ObSEArray<int64_t, 8> micro_ckpt_entry_list;
ASSERT_EQ(OB_SUCCESS, micro_ckpt_entry_list.assign(cur_super_blk.micro_ckpt_entry_list_));
int64_t micro_ckpt_list_cnt = micro_ckpt_entry_list.count();
// 4.6. finish checkpoint
ASSERT_EQ(OB_SUCCESS, phy_blk_mgr->update_block_gc_reuse_version(reuse_version_arr));
int64_t ori_reusable_cnt = phy_blk_mgr->get_reusable_set_count();
ASSERT_EQ(OB_SUCCESS, micro_ckpt_task.ckpt_op_.add_into_reusable_set(
micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.prev_super_block_.micro_ckpt_entry_list_));
ASSERT_EQ(ori_reusable_cnt, phy_blk_mgr->get_reusable_set_count());
micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.exe_round_ = 0;
// 4.7. read micro_block checkpoint, and check it
micro_meta_mgr->micro_meta_map_.clear();
ASSERT_EQ(OB_SUCCESS, micro_cache->read_micro_meta_checkpoint(cur_super_blk.micro_ckpt_entry_list_.at(0),
cur_super_blk.micro_ckpt_time_us_));
ASSERT_EQ(ori_arc_limit, micro_meta_mgr->get_arc_info().limit_);
ASSERT_EQ((WRITE_BLK_CNT - 1) * micro_cnt, micro_meta_mgr->replay_ctx_.total_replay_cnt_);
ASSERT_EQ((WRITE_BLK_CNT - 2) * micro_cnt, micro_meta_mgr->arc_info_.seg_info_arr_[ARC_T1].cnt_);
ASSERT_EQ(micro_cnt, micro_meta_mgr->arc_info_.seg_info_arr_[ARC_B1].cnt_);
// 5. execute phy_block checkpoint
// 5.1. check ckpt state
blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.exe_round_ = ObSSExecuteBlkCheckpointOp::BLK_INFO_CKPT_INTERVAL_ROUND;
blk_ckpt_task.is_inited_ = true;
ASSERT_EQ(OB_SUCCESS, blk_ckpt_task.ckpt_op_.check_state());
ASSERT_EQ(OB_SUCCESS, phy_blk_mgr->get_ss_super_block(blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.prev_super_block_));
ASSERT_EQ(true, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.need_ckpt_);
ASSERT_EQ(0, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.cur_super_block_.micro_ckpt_entry_list_.count());
ASSERT_EQ(0, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.cur_super_block_.blk_ckpt_entry_list_.count());
ASSERT_EQ(blk_ckpt_list_cnt, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.prev_super_block_.blk_ckpt_entry_list_.count());
ASSERT_EQ(micro_ckpt_list_cnt, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.prev_super_block_.micro_ckpt_entry_list_.count());
// 5.2. gen phy_blk checkpoint
ASSERT_EQ(OB_SUCCESS, blk_ckpt_task.ckpt_op_.gen_phy_block_checkpoint());
ASSERT_EQ(blk_ckpt_list_cnt, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.cur_super_block_.blk_ckpt_entry_list_.count());
const int64_t total_blk_cnt = phy_blk_mgr->blk_cnt_info_.total_blk_cnt_ - 2;
ASSERT_EQ(total_blk_cnt, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.ckpt_item_cnt_);
// 5.3. update ss_super_block
ASSERT_EQ(OB_SUCCESS, blk_ckpt_task.ckpt_op_.update_super_block(false));
cur_super_blk.reset();
ASSERT_EQ(OB_SUCCESS, phy_blk_mgr->get_ss_super_block(cur_super_blk));
ASSERT_EQ(micro_ckpt_list_cnt, cur_super_blk.micro_ckpt_entry_list_.count());
ASSERT_EQ(blk_ckpt_list_cnt, cur_super_blk.blk_ckpt_entry_list_.count());
ASSERT_NE(blk_ckpt_entry_list.at(0), cur_super_blk.blk_ckpt_entry_list_.at(0));
blk_ckpt_entry_list.reset();
ASSERT_EQ(OB_SUCCESS, blk_ckpt_entry_list.assign(cur_super_blk.blk_ckpt_entry_list_));
for (int64_t i = 0; i < micro_ckpt_entry_list.count(); ++i) {
ASSERT_EQ(micro_ckpt_entry_list.at(i), cur_super_blk.micro_ckpt_entry_list_.at(i));
}
// 5.4. finish checkpoint(evicted phy_block and invalid phy_block have already been reused)
ObSSPhysicalBlockHandle phy_blk_handle;
ASSERT_EQ(OB_SUCCESS, phy_blk_mgr->get_block_handle(evict_blk_idx, phy_blk_handle));
ASSERT_EQ(true, phy_blk_handle.is_valid());
uint64_t ori_reuse_version_1 = phy_blk_handle.get_ptr()->reuse_version_;
phy_blk_handle.reset();
ASSERT_EQ(OB_SUCCESS, phy_blk_mgr->get_block_handle(invalid_blk_idx, phy_blk_handle));
ASSERT_EQ(true, phy_blk_handle.is_valid());
uint64_t ori_reuse_version_2 = phy_blk_handle.get_ptr()->reuse_version_;
phy_blk_handle.reset();
ASSERT_EQ(OB_SUCCESS, blk_ckpt_task.ckpt_op_.try_free_phy_block(blk_ckpt_task.ckpt_op_.reusable_block_idxs_));
ASSERT_EQ(OB_SUCCESS, phy_blk_mgr->get_block_handle(evict_blk_idx, phy_blk_handle));
ASSERT_EQ(true, phy_blk_handle.is_valid());
ASSERT_EQ(true, phy_blk_handle.ptr_->is_free_);
ASSERT_EQ(ori_reuse_version_1 + 1, phy_blk_handle.ptr_->reuse_version_);
ASSERT_EQ(ori_reuse_version_1, phy_blk_handle.ptr_->gc_reuse_version_);
phy_blk_handle.reset();
ASSERT_EQ(OB_SUCCESS, phy_blk_mgr->get_block_handle(invalid_blk_idx, phy_blk_handle));
ASSERT_EQ(true, phy_blk_handle.is_valid());
ASSERT_EQ(true, phy_blk_handle.ptr_->is_free_);
ASSERT_EQ(ori_reuse_version_2 + 1, phy_blk_handle.ptr_->reuse_version_);
ASSERT_EQ(ori_reuse_version_2, phy_blk_handle.ptr_->gc_reuse_version_);
phy_blk_handle.reset();
ASSERT_EQ(phy_blk_mgr->blk_cnt_info_.micro_ckpt_blk_.used_cnt_, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.cur_super_block_.micro_ckpt_entry_list_.count());
ASSERT_EQ(phy_blk_mgr->blk_cnt_info_.phy_ckpt_blk_.used_cnt_, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.cur_super_block_.blk_ckpt_entry_list_.count());
// 5.5. read phy_block checkpoint, and check it
ObArray<ObSSPhyBlockReuseInfo> reuse_info_arr1;
ASSERT_EQ(OB_SUCCESS, phy_blk_mgr->scan_blocks_to_ckpt(reuse_info_arr1));
ASSERT_LT(0, reuse_info_arr1.count());
ASSERT_EQ(OB_SUCCESS, micro_cache->read_phy_block_checkpoint(cur_super_blk.blk_ckpt_entry_list_.at(0)));
ObArray<ObSSPhyBlockReuseInfo> reuse_info_arr2;
ASSERT_EQ(OB_SUCCESS, phy_blk_mgr->scan_blocks_to_ckpt(reuse_info_arr2));
ASSERT_EQ(reuse_info_arr1.count(), reuse_info_arr2.count());
for (int64_t i = 2; i < reuse_info_arr1.count(); ++i) {
ASSERT_EQ(reuse_info_arr1.at(i).blk_idx_, reuse_info_arr2.at(i).blk_idx_);
ASSERT_EQ(reuse_info_arr1.at(i).reuse_version_, reuse_info_arr2.at(i).reuse_version_);
}
// 6. execute phy_block checkpoint
blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.reuse();
blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.need_ckpt_ = true;
ASSERT_EQ(false, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.prev_super_block_.is_valid());
ASSERT_EQ(false, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.cur_super_block_.is_valid());
ASSERT_EQ(0, phy_blk_mgr->reusable_set_.size());
ASSERT_EQ(OB_SUCCESS, blk_ckpt_task.ckpt_op_.gen_checkpoint());
ASSERT_EQ(true, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.prev_super_block_.is_valid());
ASSERT_EQ(true, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.cur_super_block_.is_valid());
ASSERT_EQ(phy_blk_mgr->blk_cnt_info_.total_blk_cnt_ - 2, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.ckpt_item_cnt_);
ASSERT_EQ(blk_ckpt_list_cnt, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.prev_super_block_.blk_ckpt_entry_list_.count());
ASSERT_EQ(micro_ckpt_list_cnt, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.prev_super_block_.micro_ckpt_entry_list_.count());
// 7. execute micro_meta checkpoint
micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.reuse();
micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.need_ckpt_ = true;
ASSERT_EQ(false, micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.prev_super_block_.is_valid());
ASSERT_EQ(false, micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.cur_super_block_.is_valid());
ASSERT_EQ(OB_SUCCESS, phy_blk_mgr->scan_blocks_to_reuse());
ASSERT_EQ(0, phy_blk_mgr->reusable_set_.size());
ASSERT_EQ(OB_SUCCESS, micro_ckpt_task.ckpt_op_.gen_checkpoint());
ASSERT_EQ(true, micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.prev_super_block_.is_valid());
ASSERT_EQ(true, micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.cur_super_block_.is_valid());
ASSERT_EQ((WRITE_BLK_CNT - 1) * micro_cnt, micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.ckpt_item_cnt_);
ASSERT_EQ(blk_ckpt_list_cnt, micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.prev_super_block_.blk_ckpt_entry_list_.count());
ASSERT_EQ(micro_ckpt_entry_list.count(), micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.prev_super_block_.micro_ckpt_entry_list_.count());
ASSERT_EQ(phy_blk_mgr->blk_cnt_info_.micro_ckpt_blk_.used_cnt_, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.cur_super_block_.micro_ckpt_entry_list_.count() * 2);
ASSERT_EQ(phy_blk_mgr->blk_cnt_info_.phy_ckpt_blk_.used_cnt_, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.cur_super_block_.blk_ckpt_entry_list_.count());
// 8. execute phy_block checkpoint
blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.reuse();
blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.need_ckpt_ = true;
ASSERT_EQ(false, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.prev_super_block_.is_valid());
ASSERT_EQ(false, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.cur_super_block_.is_valid());
ASSERT_EQ(micro_ckpt_entry_list.count(), phy_blk_mgr->reusable_set_.size());
ASSERT_EQ(OB_SUCCESS, blk_ckpt_task.ckpt_op_.gen_checkpoint());
ASSERT_EQ(true, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.prev_super_block_.is_valid());
ASSERT_EQ(true, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.cur_super_block_.is_valid());
ASSERT_EQ(phy_blk_mgr->blk_cnt_info_.total_blk_cnt_ - 2, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.ckpt_item_cnt_);
ASSERT_EQ(blk_ckpt_list_cnt, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.prev_super_block_.blk_ckpt_entry_list_.count());
ASSERT_EQ(micro_ckpt_list_cnt, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.prev_super_block_.micro_ckpt_entry_list_.count());
ASSERT_EQ(phy_blk_mgr->blk_cnt_info_.micro_ckpt_blk_.used_cnt_, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.cur_super_block_.micro_ckpt_entry_list_.count());
ASSERT_EQ(phy_blk_mgr->blk_cnt_info_.phy_ckpt_blk_.used_cnt_, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.cur_super_block_.blk_ckpt_entry_list_.count());
allocator.clear();
}
/* This case tests whether the micro cache can be restored to the expected state after restart. */
TEST_F(TestSSExecuteCheckpointTask, test_micro_cache_ckpt_after_restart)
{
ObSSMicroCacheStat &cache_stat = micro_cache_->cache_stat_;
micro_ckpt_task_->is_inited_ = false;
blk_ckpt_task_->is_inited_ = false;
arc_task_->is_inited_ = false;
persist_task_->is_inited_ = false;
const int64_t total_blk_cnt = phy_blk_mgr_->blk_cnt_info_.total_blk_cnt_;
ob_usleep(1000 * 1000);
// 1. add some micro_block into cache and record their micro_meta
ObArray<ObSSMicroBlockMeta *> micro_meta_arr1;
int64_t macro_start_idx = 1;
int64_t write_blk_cnt = 10;
ASSERT_EQ(OB_SUCCESS, add_batch_micro_block(macro_start_idx, write_blk_cnt, micro_meta_arr1));
const int64_t used_normal_blk_cnt1 = phy_blk_mgr_->blk_cnt_info_.normal_blk_.used_cnt_;
ObArray<ObSSPhyBlockPersistInfo> phy_block_arr1;
int64_t alloc_phy_blk_cnt = 20;
ASSERT_EQ(OB_SUCCESS, alloc_batch_phy_block_to_reuse(alloc_phy_blk_cnt, phy_block_arr1));
// 2. do ckpt_task first round
micro_ckpt_task_->is_inited_ = true;
blk_ckpt_task_->is_inited_ = true;
micro_ckpt_task_->interval_us_ = 3600 * 1000 * 1000L;
blk_ckpt_task_->interval_us_ = 3600 * 1000 * 1000L;
ob_usleep(1000 * 1000);
micro_ckpt_task_->ckpt_op_.micro_ckpt_ctx_.need_ckpt_ = true;
micro_ckpt_task_->ckpt_op_.micro_ckpt_ctx_.need_scan_blk_ = true;
blk_ckpt_task_->ckpt_op_.blk_ckpt_ctx_.need_ckpt_ = true;
ASSERT_EQ(false, micro_ckpt_task_->ckpt_op_.micro_ckpt_ctx_.prev_super_block_.is_valid());
ASSERT_EQ(false, micro_ckpt_task_->ckpt_op_.micro_ckpt_ctx_.cur_super_block_.is_valid());
ASSERT_EQ(false, blk_ckpt_task_->ckpt_op_.blk_ckpt_ctx_.prev_super_block_.is_valid());
ASSERT_EQ(false, blk_ckpt_task_->ckpt_op_.blk_ckpt_ctx_.cur_super_block_.is_valid());
ASSERT_EQ(0, phy_blk_mgr_->blk_cnt_info_.micro_ckpt_blk_.used_cnt_);
ASSERT_EQ(0, phy_blk_mgr_->blk_cnt_info_.phy_ckpt_blk_.used_cnt_);
ASSERT_EQ(0, phy_blk_mgr_->reusable_set_.size());
int64_t start_time_us = ObTimeUtility::current_time();
ASSERT_EQ(OB_SUCCESS, micro_ckpt_task_->ckpt_op_.gen_checkpoint());
const int64_t micro_exe_time_us = ObTimeUtility::current_time() - start_time_us;
ASSERT_LT(0, phy_blk_mgr_->blk_cnt_info_.micro_ckpt_blk_.used_cnt_);
ASSERT_LT(0, phy_blk_mgr_->reusable_set_.size());
ASSERT_EQ((write_blk_cnt - 1) * 20, micro_ckpt_task_->ckpt_op_.micro_ckpt_ctx_.ckpt_item_cnt_);
start_time_us = ObTimeUtility::current_time();
ASSERT_EQ(OB_SUCCESS, blk_ckpt_task_->ckpt_op_.gen_checkpoint());
const int64_t blk_exe_time_us = ObTimeUtility::current_time() - start_time_us;
ASSERT_LT(0, phy_blk_mgr_->blk_cnt_info_.phy_ckpt_blk_.used_cnt_);
ASSERT_EQ(total_blk_cnt - SS_CACHE_SUPER_BLOCK_CNT, blk_ckpt_task_->ckpt_op_.blk_ckpt_ctx_.ckpt_item_cnt_);
micro_ckpt_task_->is_inited_ = false;
blk_ckpt_task_->is_inited_ = false;
ObSSMicroCacheSuperBlock super_block1 = phy_blk_mgr_->super_block_;
// 3. restart micro cache
LOG_INFO("TEST: start first restart");
restart_micro_cache();
LOG_INFO("TEST: finish first restart");
// 4. check super_block, micro_meta and phy_block_info
check_phy_blk_info(phy_block_arr1, 1);
check_micro_meta(micro_meta_arr1);
check_super_block(super_block1);
ASSERT_EQ(super_block1.micro_ckpt_entry_list_.count(), phy_blk_mgr_->blk_cnt_info_.micro_ckpt_blk_.used_cnt_);
ASSERT_EQ(super_block1.blk_ckpt_entry_list_.count(), phy_blk_mgr_->blk_cnt_info_.phy_ckpt_blk_.used_cnt_);
ASSERT_EQ(used_normal_blk_cnt1, phy_blk_mgr_->blk_cnt_info_.normal_blk_.used_cnt_);
// 5. clear all micro meta
micro_meta_mgr_->micro_meta_map_.reset();
// 6. repeat step 1
ObArray<ObSSMicroBlockMeta *> micro_meta_arr2;
macro_start_idx += write_blk_cnt;
write_blk_cnt = 10;
ASSERT_EQ(OB_SUCCESS, add_batch_micro_block(macro_start_idx, write_blk_cnt, micro_meta_arr2));
const int64_t used_normal_blk_cnt2 = phy_blk_mgr_->blk_cnt_info_.normal_blk_.used_cnt_;
alloc_phy_blk_cnt = 20;
ObArray<ObSSPhyBlockPersistInfo> phy_block_arr2;
ASSERT_EQ(OB_SUCCESS, alloc_batch_phy_block_to_reuse(alloc_phy_blk_cnt, phy_block_arr2));
// 7. do ckpt_task second round and randomly force to end ckpt_task and restart micro_cache
micro_ckpt_task_->is_inited_ = true;
blk_ckpt_task_->is_inited_ = true;
micro_ckpt_task_->interval_us_ = 3600 * 1000 * 1000L;
blk_ckpt_task_->interval_us_ = 3600 * 1000 * 1000L;
ob_usleep(1000 * 1000);
micro_ckpt_task_->ckpt_op_.micro_ckpt_ctx_.need_ckpt_ = true;
micro_ckpt_task_->ckpt_op_.micro_ckpt_ctx_.need_scan_blk_ = true;
blk_ckpt_task_->ckpt_op_.blk_ckpt_ctx_.need_ckpt_ = true;
std::thread micro_t([&]() {
ObRandom rand;
const int64_t sleep_us = ObRandom::rand(1, micro_exe_time_us * 2);
ob_usleep(sleep_us);
micro_ckpt_task_->ckpt_op_.is_inited_ = false; // force to end micro_ckpt
});
std::thread blk_t([&]() {
ObRandom rand;
const int64_t sleep_us = ObRandom::rand(1, blk_exe_time_us * 2);
ob_usleep(sleep_us);
blk_ckpt_task_->ckpt_op_.is_inited_ = false; // force to end blk_ckpt
});
blk_ckpt_task_->ckpt_op_.gen_checkpoint();
micro_ckpt_task_->ckpt_op_.gen_checkpoint();
ObSSMicroCacheSuperBlock super_block2 = phy_blk_mgr_->super_block_;
blk_t.join();
micro_t.join();
LOG_INFO("TEST: start second restart");
restart_micro_cache();
LOG_INFO("TEST: finish second restart");
// // 8. check super_block, micro_meta and phy_block_info
// if (phy_blk_mgr_->super_block_.modify_time_us_ == super_block1.modify_time_us_) {
// // fail to generate new ckpt
// LOG_INFO("TEST: only ckpt once", K(super_block1));
// check_phy_blk_info(phy_block_arr2, 0);
// check_micro_meta(micro_meta_arr1);
// check_super_block(super_block1);
// ASSERT_EQ(super_block1.micro_ckpt_entry_list_.count(), phy_blk_mgr_->blk_cnt_info_.micro_ckpt_blk_.used_cnt_);
// ASSERT_EQ(super_block1.blk_ckpt_entry_list_.count(), phy_blk_mgr_->blk_cnt_info_.phy_ckpt_blk_.used_cnt_);
// ASSERT_EQ(used_normal_blk_cnt1, phy_blk_mgr_->blk_cnt_info_.normal_blk_.used_cnt_);
// } else {
// ASSERT_GT(phy_blk_mgr_->super_block_.modify_time_us_, super_block1.modify_time_us_);
// // succeed to generate new ckpt
// LOG_INFO("TEST: ckpt twice", K(super_block2));
// check_phy_blk_info(phy_block_arr2, 1);
// check_micro_meta(micro_meta_arr2);
// check_super_block(super_block2);
// ASSERT_EQ(super_block2.micro_ckpt_entry_list_.count(), phy_blk_mgr_->blk_cnt_info_.micro_ckpt_blk_.used_cnt_);
// ASSERT_EQ(super_block2.blk_ckpt_entry_list_.count(), phy_blk_mgr_->blk_cnt_info_.phy_ckpt_blk_.used_cnt_);
// // Because the micro_meta written by round1 has been cleared, if round2 do ckpt_task successfully,
// // after micro_cache restart, the phy_block corresponding to the micro_block written by round1 will be recycled.
// ASSERT_GT(used_normal_blk_cnt2, phy_blk_mgr_->blk_cnt_info_.normal_blk_.used_cnt_);
// }
}
/* After micro_ckpt_task execucte scan_blocks_to_reuse, must set need_scan_blk_ = false */
TEST_F(TestSSExecuteCheckpointTask, test_micro_ckpt_task_exec_scan_block)
{
int ret = OB_SUCCESS;
ObSSMicroCache *micro_cache = MTL(ObSSMicroCache *);
ASSERT_NE(nullptr, micro_cache);
ObSSPhysicalBlockManager &phy_blk_mgr = micro_cache->phy_blk_mgr_;
ObSSExecuteMicroCheckpointTask &micro_ckpt_task = micro_cache->task_runner_.micro_ckpt_task_;
ObSSExecuteBlkCheckpointTask &blk_ckpt_task = micro_cache->task_runner_.blk_ckpt_task_;
blk_ckpt_task.is_inited_ = false;
const int64_t block_cnt = 10;
for (int64_t i = 0; i < block_cnt; i++) {
int64_t phy_blk_idx = -1;
ObSSPhysicalBlockHandle phy_blk_handle;
ASSERT_EQ(OB_SUCCESS, phy_blk_mgr.alloc_block(phy_blk_idx, phy_blk_handle, ObSSPhyBlockType::SS_NORMAL_BLK));
phy_blk_handle()->is_free_ = false;
phy_blk_handle()->is_sealed_ = true;
phy_blk_handle()->valid_len_ = 0;
}
micro_ckpt_task.ckpt_op_.ckpt_ctx_->exe_round_ = ObSSExecuteMicroCheckpointOp::SCAN_BLOCK_INTERVAL_ROUND - 1;
ob_usleep(3 * 1000 * 1000);
ASSERT_EQ(block_cnt, phy_blk_mgr.get_reusable_set_count());
ASSERT_EQ(false, micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.need_scan_blk_);
}
} // namespace storage
} // namespace oceanbase
int main(int argc, char **argv)
{
system("rm -f test_ss_execute_checkpoint_task.log*");
OB_LOGGER.set_file_name("test_ss_execute_checkpoint_task.log", true, true);
OB_LOGGER.set_log_level("INFO");
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}
Reference in New Issue View Git Blame Copy Permalink