hcq/oceanbase
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity
Files
develop
oceanbase/mittest/shared_storage/test_ss_execute_checkpoint_task.cpp

1000 lines
50 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 "mittest/mtlenv/mock_tenant_module_env.h"
#include "test_ss_common_util.h"
#include "storage/shared_storage/micro_cache/ckpt/ob_ss_linked_phy_block_reader.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 macro_cnt, const int64_t micro_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, const bool is_exist);
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*);
// clear micro_meta_ckpt
ObSSMicroCacheSuperBlock new_super_blk(micro_cache->cache_file_size_);
ASSERT_EQ(OB_SUCCESS, micro_cache->phy_blk_mgr_.update_ss_super_block(new_super_blk));
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 macro_cnt,
const int64_t micro_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_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 || macro_cnt <= 0 || micro_cnt <= 0)) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", KR(ret), K(start_idx), K(macro_cnt), K(micro_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 + macro_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_CACHE_DATA_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,
const bool is_exist)
{
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);
if (is_exist == true) {
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);
}
} 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_.meta_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->blk_cnt_info_.cache_limit_blk_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_.data_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;
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));
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;
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));
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 reserved_micro_ckpt_blk were used up, will fail to gen micro_meta ckpt
const int64_t ori_micro_blk_used_cnt = phy_blk_mgr->blk_cnt_info_.meta_blk_.used_cnt_;
const int64_t ori_micro_blk_hold_cnt = phy_blk_mgr->blk_cnt_info_.meta_blk_.hold_cnt_;
const int64_t ori_micro_blk_max_cnt = phy_blk_mgr->blk_cnt_info_.meta_blk_.max_cnt_;
phy_blk_mgr->blk_cnt_info_.meta_blk_.used_cnt_ = ori_micro_blk_max_cnt;
phy_blk_mgr->blk_cnt_info_.meta_blk_.hold_cnt_ = ori_micro_blk_max_cnt;
phy_blk_mgr->blk_cnt_info_.shared_blk_used_cnt_ =
phy_blk_mgr->blk_cnt_info_.meta_blk_.hold_cnt_ + phy_blk_mgr->blk_cnt_info_.data_blk_.hold_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_.meta_blk_.used_cnt_ = ori_micro_blk_used_cnt;
phy_blk_mgr->blk_cnt_info_.meta_blk_.hold_cnt_ = ori_micro_blk_hold_cnt;
phy_blk_mgr->blk_cnt_info_.shared_blk_used_cnt_ =
phy_blk_mgr->blk_cnt_info_.meta_blk_.hold_cnt_ + phy_blk_mgr->blk_cnt_info_.data_blk_.hold_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 free_blk_cnt = 0;
ASSERT_EQ(OB_SUCCESS,
phy_blk_mgr->try_free_batch_blocks(micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.prev_super_block_.micro_ckpt_entry_list_, free_blk_cnt));
ASSERT_EQ(free_blk_cnt, micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.prev_super_block_.micro_ckpt_entry_list_.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,
phy_blk_mgr->try_free_batch_blocks(blk_ckpt_task.ckpt_op_.reusable_block_idxs_, blk_ckpt_task.ckpt_op_.blk_ckpt_ctx_.free_blk_cnt_));
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_.meta_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_.meta_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());
// 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(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_.meta_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;
const int64_t micro_cnt = 20;
ASSERT_EQ(OB_SUCCESS, add_batch_micro_block(macro_start_idx, write_blk_cnt, micro_cnt, micro_meta_arr1));
const int64_t used_data_blk_cnt1 = phy_blk_mgr_->blk_cnt_info_.data_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_.meta_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_.meta_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, true);
check_super_block(super_block1);
ASSERT_EQ(super_block1.micro_ckpt_entry_list_.count(), phy_blk_mgr_->blk_cnt_info_.meta_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_data_blk_cnt1, phy_blk_mgr_->blk_cnt_info_.data_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 = 20;
ASSERT_EQ(OB_SUCCESS, add_batch_micro_block(macro_start_idx, write_blk_cnt, micro_cnt, micro_meta_arr2));
const int64_t used_data_blk_cnt2 = phy_blk_mgr_->blk_cnt_info_.data_blk_.used_cnt_;
// 7. do ckpt_task second round and randomly force to end micro_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;
ObSSMicroCacheSuperBlock super_block2 = phy_blk_mgr_->super_block_;
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
});
micro_ckpt_task_->ckpt_op_.gen_checkpoint();
micro_t.join();
ObSSMicroCacheSuperBlock super_block3 = phy_blk_mgr_->super_block_;
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_block2.modify_time_us_) {
check_super_block(super_block2);
check_micro_meta(micro_meta_arr1, true);
check_micro_meta(micro_meta_arr2, false);
ASSERT_EQ(used_data_blk_cnt1, phy_blk_mgr_->blk_cnt_info_.data_blk_.used_cnt_);
} else {
check_super_block(super_block3);
check_micro_meta(micro_meta_arr1, false);
check_micro_meta(micro_meta_arr2, true);
ASSERT_EQ(used_data_blk_cnt2 - used_data_blk_cnt1, phy_blk_mgr_->blk_cnt_info_.data_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)
{
LOG_INFO("TEST: 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_CACHE_DATA_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_);
}
/* Test whether extra meta blocks can be dynamically allocated when the number of blocks required
* by micro_ckpt exceeds meta_blk.min_cnt_ */
TEST_F(TestSSExecuteCheckpointTask, test_reserve_micro_ckpt_blk)
{
LOG_INFO("TEST_CASE: start test_reserve_micro_ckpt_blk");
int ret = OB_SUCCESS;
ObSSMicroCache *micro_cache = MTL(ObSSMicroCache *);
ASSERT_NE(nullptr, micro_cache);
ObSSMicroCacheStat &cache_stat = micro_cache->cache_stat_;
ObSSExecuteMicroCheckpointTask &micro_ckpt_task = micro_cache->task_runner_.micro_ckpt_task_;
ObSSPhysicalBlockManager &phy_blk_mgr = micro_cache->phy_blk_mgr_;
SSPhyBlockCntInfo &blk_cnt_info = micro_cache->phy_blk_mgr_.blk_cnt_info_;
// 1. write enough micro blocks so that the number of blocks required to execute micro_ckpt exceeds meta_blk.min_cnt_
const int64_t extra_meta_blk_cnt = 5;
const int64_t estimate_micro_cnt =
(blk_cnt_info.meta_blk_.hold_cnt_ + extra_meta_blk_cnt) * phy_blk_mgr.get_block_size() / AVG_MICRO_META_PERSIST_COST;
const int64_t macro_cnt = blk_cnt_info.data_blk_.hold_cnt_ * 0.2;
const int64_t micro_cnt = estimate_micro_cnt / macro_cnt;
ObArray<ObSSMicroBlockMeta *> micro_meta_arr;
ASSERT_EQ(OB_SUCCESS, add_batch_micro_block(1, macro_cnt, micro_cnt, micro_meta_arr));
// 2. mock shared_block is used up
blk_cnt_info.data_blk_.hold_cnt_ = blk_cnt_info.data_blk_.max_cnt_;
blk_cnt_info.shared_blk_used_cnt_ = blk_cnt_info.data_blk_.hold_cnt_ + blk_cnt_info.meta_blk_.hold_cnt_;
// 3. execute micro_meta_ckpt
ASSERT_EQ(0, blk_cnt_info.meta_blk_.used_cnt_);
micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.exe_round_ = ObSSExecuteMicroCheckpointOp::MICRO_META_CKPT_INTERVAL_ROUND - 1;
ob_usleep(10 * 1000 * 1000);
ASSERT_EQ(blk_cnt_info.meta_blk_.hold_cnt_, blk_cnt_info.meta_blk_.used_cnt_);
ASSERT_EQ(blk_cnt_info.meta_blk_.used_cnt_, phy_blk_mgr.super_block_.micro_ckpt_entry_list_.count());
// 4. revert step2
blk_cnt_info.data_blk_.hold_cnt_ = blk_cnt_info.data_blk_.min_cnt_;
blk_cnt_info.shared_blk_used_cnt_ = blk_cnt_info.data_blk_.hold_cnt_ + blk_cnt_info.meta_blk_.hold_cnt_;
// 5. execute micro_meta_ckpt again, check that the number of blocks used by micro_ckpt has increased
const int64_t origin_meta_blk_usd_cnt = blk_cnt_info.meta_blk_.used_cnt_;
micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.exe_round_ = ObSSExecuteMicroCheckpointOp::ESTIMATE_MICRO_CKPT_BLK_CNT_ROUND - 1;
ob_usleep(2 * 1000 * 1000);
ASSERT_LT(2 * origin_meta_blk_usd_cnt, blk_cnt_info.meta_blk_.hold_cnt_);
micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.exe_round_ = ObSSExecuteMicroCheckpointOp::MICRO_META_CKPT_INTERVAL_ROUND - 1;
ob_usleep(10 * 1000 * 1000);
ASSERT_EQ(blk_cnt_info.meta_blk_.used_cnt_, phy_blk_mgr.super_block_.micro_ckpt_entry_list_.count());
ASSERT_LT(origin_meta_blk_usd_cnt, blk_cnt_info.meta_blk_.used_cnt_);
ASSERT_LT(origin_meta_blk_usd_cnt, phy_blk_mgr.super_block_.micro_ckpt_entry_list_.count());
}
TEST_F(TestSSExecuteCheckpointTask, test_dynamic_update_arc_limit)
{
LOG_INFO("TEST_CASE: start test_dynamic_update_arc_limit");
ObSSMicroCache *micro_cache = MTL(ObSSMicroCache *);
ObSSARCInfo &arc_info = micro_cache->micro_meta_mgr_.arc_info_;
ObSSPhysicalBlockManager &phy_blk_mgr = micro_cache->phy_blk_mgr_;
SSPhyBlockCntInfo &blk_cnt_info = phy_blk_mgr.blk_cnt_info_;
ObSSExecuteMicroCheckpointOp &micro_ckpt_op = micro_cache->task_runner_.micro_ckpt_task_.ckpt_op_;
const int64_t origin_limit = arc_info.limit_;
const int64_t origin_work_limit = arc_info.work_limit_;
const int64_t origin_p = arc_info.p_;
const int64_t origin_max_p = arc_info.max_p_;
const int64_t origin_min_p = arc_info.min_p_;
// mock micro_ckpt use extra block from shared_blocks.
// this will result in fewer available cache_data_blocks and dynamically lower arc_limit.
blk_cnt_info.meta_blk_.hold_cnt_ = blk_cnt_info.meta_blk_.max_cnt_;
blk_cnt_info.shared_blk_used_cnt_ = blk_cnt_info.meta_blk_.hold_cnt_ + blk_cnt_info.data_blk_.hold_cnt_;
micro_ckpt_op.dynamic_update_arc_limit();
ASSERT_GT(origin_limit, arc_info.limit_);
ASSERT_GT(origin_work_limit, arc_info.work_limit_);
ASSERT_GT(origin_p, arc_info.p_);
ASSERT_GT(origin_max_p, arc_info.max_p_);
ASSERT_GT(origin_min_p, arc_info.min_p_);
blk_cnt_info.meta_blk_.hold_cnt_ = blk_cnt_info.meta_blk_.min_cnt_;
blk_cnt_info.shared_blk_used_cnt_ = blk_cnt_info.meta_blk_.hold_cnt_ + blk_cnt_info.data_blk_.hold_cnt_;
micro_ckpt_op.dynamic_update_arc_limit();
ASSERT_EQ(origin_limit, arc_info.limit_);
ASSERT_EQ(origin_work_limit, arc_info.work_limit_);
ASSERT_LE(abs(origin_p - arc_info.p_), 5); // the conversion between 'int64_t' and 'double' causes some deviations
ASSERT_EQ(origin_max_p, arc_info.max_p_);
ASSERT_EQ(origin_min_p, arc_info.min_p_);
// test dynamically update arc_limit when prewarming
micro_cache->begin_free_space_for_prewarm();
const int64_t prewarm_work_limit = static_cast<int64_t>((static_cast<double>(origin_work_limit * SS_ARC_LIMIT_SHRINK_PCT) / 100.0));
double ori_pct = static_cast<double>(origin_p * 100) / origin_work_limit;
const int64_t p1 = static_cast<int64_t>((static_cast<double>(ori_pct * prewarm_work_limit) / 100.0));
ASSERT_EQ(prewarm_work_limit, arc_info.work_limit_);
ASSERT_EQ(origin_limit, arc_info.limit_);
ASSERT_LE(abs(p1 - arc_info.p_), 5); // the conversion between 'int64_t' and 'double' causes some deviations
// mock micro_ckpt use extra block from shared_blocks.
// this will result in fewer available cache_data_blocks and dynamically lower arc_limit.
blk_cnt_info.meta_blk_.hold_cnt_ = blk_cnt_info.meta_blk_.max_cnt_;
blk_cnt_info.shared_blk_used_cnt_ = blk_cnt_info.meta_blk_.hold_cnt_ + blk_cnt_info.data_blk_.hold_cnt_;
micro_ckpt_op.dynamic_update_arc_limit();
const int64_t new_limit = phy_blk_mgr.get_cache_limit_size() * SS_ARC_LIMIT_PCT / 100;
const int64_t new_work_limit = (static_cast<double>(prewarm_work_limit) / origin_limit) * new_limit;
ori_pct = static_cast<double>(p1 * 100) / prewarm_work_limit;
const int64_t p2 = static_cast<int64_t>((static_cast<double>(new_work_limit * ori_pct) / 100.0));
ASSERT_EQ(new_limit, arc_info.limit_);
ASSERT_EQ(new_work_limit, arc_info.work_limit_);
ASSERT_LE(abs(p2 - arc_info.p_), 5); // the conversion between 'int64_t' and 'double' causes some deviations
micro_cache->finish_free_space_for_prewarm();
ori_pct = static_cast<double>(p2 * 100) / new_work_limit;
const int64_t p3 = static_cast<int64_t>((static_cast<double>(new_limit * ori_pct) / 100.0));
ASSERT_EQ(new_limit, arc_info.work_limit_);
ASSERT_EQ(new_limit, arc_info.limit_);
ASSERT_LE(abs(p3 - arc_info.p_), 5); // the conversion between 'int64_t' and 'double' causes some deviations
}
TEST_F(TestSSExecuteCheckpointTask, test_estimate_micro_meta_persist_cost)
{
LOG_INFO("TEST_CASE: start test_estimate_micro_meta_persist_cost");
int ret = OB_SUCCESS;
ObSSMicroCache *micro_cache = MTL(ObSSMicroCache *);
ASSERT_NE(nullptr, micro_cache);
ObSSExecuteMicroCheckpointTask &micro_ckpt_task = micro_cache->task_runner_.micro_ckpt_task_;
ObSSPhysicalBlockManager &phy_blk_mgr = micro_cache->phy_blk_mgr_;
SSPhyBlockCntInfo &blk_cnt_info = micro_cache->phy_blk_mgr_.blk_cnt_info_;
const int64_t estimate_micro_cnt =
(blk_cnt_info.meta_blk_.hold_cnt_) * phy_blk_mgr.get_block_size() / AVG_MICRO_META_PERSIST_COST;
const int64_t macro_cnt = blk_cnt_info.data_blk_.hold_cnt_ * 0.2;
const int64_t micro_cnt = estimate_micro_cnt / macro_cnt;
ObArray<ObSSMicroBlockMeta *> micro_meta_arr;
ASSERT_EQ(OB_SUCCESS, add_batch_micro_block(1, macro_cnt, micro_cnt, micro_meta_arr));
// 1. execute micro_meta_ckpt
micro_ckpt_task.ckpt_op_.micro_ckpt_ctx_.exe_round_ = ObSSExecuteMicroCheckpointOp::MICRO_META_CKPT_INTERVAL_ROUND - 1;
ob_usleep(10 * 1000 * 1000);
// 2. read ckpt_blk
ObIArray<int64_t> &meta_blk_arr = phy_blk_mgr.super_block_.micro_ckpt_entry_list_;
ASSERT_LT(0, meta_blk_arr.count());
ObSSLinkedPhyBlockReader block_reader;
const int64_t entry_block_idx = meta_blk_arr.at(0);
ASSERT_EQ(OB_SUCCESS, block_reader.init(entry_block_idx, MTL_ID(), phy_blk_mgr));
int64_t total_item_cnt = 0;
// skip first block, because it is the last written block when execute micro_ckpt
for (int64_t i = 1; i < meta_blk_arr.count(); i++) {
const int64_t block_idx = meta_blk_arr.at(i);
ASSERT_EQ(OB_SUCCESS, block_reader.inner_read_block_(block_idx));
total_item_cnt += block_reader.linked_header_.item_count_;
LOG_INFO("print block header", K(i), K(block_idx), K_(block_reader.common_header), K_(block_reader.linked_header));
}
const int64_t total_blk_cnt = meta_blk_arr.count() - 1;
double avg_cost = static_cast<double>(total_blk_cnt * phy_blk_mgr.get_block_size()) / total_item_cnt;
std::cout << "avg_meta_persist_cost: " << avg_cost << ", total_meta_blk_cnt: " << total_blk_cnt << std::endl;
}
} // 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