386 lines
15 KiB
C++
386 lines
15 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 "mittest/mtlenv/mock_tenant_module_env.h"
|
|
#include "storage/shared_storage/micro_cache/ckpt/ob_ss_linked_phy_block_struct.h"
|
|
#include "storage/shared_storage/micro_cache/ckpt/ob_ss_linked_phy_block_reader.h"
|
|
#include "storage/shared_storage/micro_cache/ckpt/ob_ss_linked_phy_block_writer.h"
|
|
#include "storage/shared_storage/micro_cache/ob_ss_micro_cache_stat.h"
|
|
|
|
namespace oceanbase
|
|
{
|
|
namespace storage
|
|
{
|
|
using namespace oceanbase::storage;
|
|
|
|
class TestSSLinkedPhyBlock : public ::testing::Test
|
|
{
|
|
public:
|
|
TestSSLinkedPhyBlock(){}
|
|
virtual ~TestSSLinkedPhyBlock() {}
|
|
virtual void SetUp() override {}
|
|
virtual void TearDown() override {}
|
|
static void SetUpTestCase()
|
|
{
|
|
GCTX.startup_mode_ = observer::ObServerMode::SHARED_STORAGE_MODE;
|
|
ASSERT_EQ(OB_SUCCESS, MockTenantModuleEnv::get_instance().init());
|
|
tnt_file_mgr_ = MTL(ObTenantFileManager *);
|
|
ASSERT_NE(nullptr, tnt_file_mgr_);
|
|
ASSERT_EQ(OB_SUCCESS, phy_blk_mgr_.init(MTL_ID(), BLOCK_SIZE * 4096, BLOCK_SIZE));
|
|
}
|
|
static void TearDownTestCase()
|
|
{
|
|
tnt_file_mgr_ = nullptr;
|
|
phy_blk_mgr_.destroy();
|
|
MockTenantModuleEnv::get_instance().destroy();
|
|
}
|
|
|
|
private:
|
|
static ObSSMicroCacheStat cache_stat_;
|
|
static ObSSPhysicalBlockManager phy_blk_mgr_;
|
|
static ObTenantFileManager *tnt_file_mgr_;
|
|
static const int64_t BLOCK_SIZE;
|
|
};
|
|
ObSSMicroCacheStat TestSSLinkedPhyBlock::cache_stat_;
|
|
ObSSPhysicalBlockManager TestSSLinkedPhyBlock::phy_blk_mgr_(cache_stat_);
|
|
ObTenantFileManager *TestSSLinkedPhyBlock::tnt_file_mgr_ = nullptr;
|
|
const int64_t TestSSLinkedPhyBlock::BLOCK_SIZE = 2 * 1024 * 1024;
|
|
|
|
struct ItemInfo
|
|
{
|
|
ItemInfo() : allocator_(), buf_(nullptr), buf_len_(0), crc_(0) {}
|
|
~ItemInfo()
|
|
{
|
|
allocator_.reset();
|
|
buf_ = nullptr;
|
|
buf_len_ = 0;
|
|
crc_ = 0;
|
|
}
|
|
int init(const int64_t buf_len);
|
|
|
|
ObArenaAllocator allocator_;
|
|
char *buf_;
|
|
int64_t buf_len_;
|
|
uint32_t crc_;
|
|
|
|
TO_STRING_KV(KP_(buf), K_(buf_len), K_(crc));
|
|
};
|
|
|
|
int ItemInfo::init(const int64_t buf_len)
|
|
{
|
|
int ret = OB_SUCCESS;
|
|
if (OB_UNLIKELY(buf_len <= 0)) {
|
|
ret = OB_INVALID_ARGUMENT;
|
|
LOG_WARN("invalid argument", K(ret), K(buf_len));
|
|
} else if (OB_ISNULL(buf_ = static_cast<char *>(allocator_.alloc(buf_len)))) {
|
|
ret = OB_ALLOCATE_MEMORY_FAILED;
|
|
LOG_WARN("failed to buf", K(ret), K(buf_len));
|
|
} else {
|
|
const int64_t rv = ObRandom::rand(1, 10);
|
|
if (rv % 2 == 0) {
|
|
for (int64_t i = 0; OB_SUCC(ret) && i < buf_len; i++) {
|
|
buf_[i] = static_cast<char>(ObRandom::rand(-128, 127));
|
|
}
|
|
} else {
|
|
memset(buf_, 'a', buf_len);
|
|
}
|
|
|
|
if (OB_SUCC(ret)) {
|
|
buf_len_ = buf_len;
|
|
crc_ = static_cast<uint32_t>(ob_crc64(buf_, buf_len));
|
|
}
|
|
}
|
|
return ret;
|
|
}
|
|
|
|
int build_items(std::vector<ItemInfo> &item_arr, const std::vector<int64_t> &item_buf_len_arr)
|
|
{
|
|
int ret = OB_SUCCESS;
|
|
if (OB_UNLIKELY(item_arr.size() != item_buf_len_arr.size())) {
|
|
ret = OB_INVALID_ARGUMENT;
|
|
LOG_WARN("invalid argument", K(ret), K(item_arr.size()), K(item_buf_len_arr.size()));
|
|
}
|
|
for (int64_t i = 0; OB_SUCC(ret) && i < item_arr.size(); i++) {
|
|
if (OB_FAIL(item_arr[i].init(item_buf_len_arr[i]))) {
|
|
LOG_WARN("fail to init item", K(ret), K(i), K(item_arr.size()), K(item_buf_len_arr[i]));
|
|
}
|
|
}
|
|
return ret;
|
|
}
|
|
|
|
int write_items(ObSSLinkedPhyBlockItemWriter &writer, const std::vector<ItemInfo> &item_arr)
|
|
{
|
|
int ret = OB_SUCCESS;
|
|
for (int64_t i = 0; OB_SUCC(ret) && i < item_arr.size(); i++) {
|
|
if (OB_FAIL(writer.write_item(item_arr[i].buf_, item_arr[i].buf_len_))) {
|
|
LOG_WARN("fail to write item", K(ret), K(i), K(item_arr[i]));
|
|
}
|
|
}
|
|
if (FAILEDx(writer.close())) {
|
|
LOG_WARN("fail to close writer", K(ret));
|
|
}
|
|
return ret;
|
|
}
|
|
|
|
int read_items(ObSSLinkedPhyBlockItemReader &reader, const std::vector<ItemInfo> &item_arr)
|
|
{
|
|
int ret = OB_SUCCESS;
|
|
char *tmp_buf = nullptr;
|
|
int64_t tmp_buf_len = 0;
|
|
int64_t item_idx = -1;
|
|
uint32_t clac_crc = 0;
|
|
|
|
std::map<uint32_t, int64_t> block_items_map;
|
|
for (int64_t i = 0; OB_SUCC(ret) && i < item_arr.size(); i++) {
|
|
block_items_map[item_arr[i].crc_] = i;
|
|
}
|
|
|
|
for (int64_t i = 0; OB_SUCC(ret) && i < item_arr.size(); i++) {
|
|
tmp_buf = nullptr;
|
|
tmp_buf_len = 0;
|
|
item_idx = -1;
|
|
clac_crc = 0;
|
|
|
|
if (OB_FAIL(reader.get_next_item(tmp_buf, tmp_buf_len))) {
|
|
LOG_WARN("fail to read item", K(ret), K(i), K(item_arr[i]));
|
|
} else if (OB_ISNULL(tmp_buf) || OB_UNLIKELY(tmp_buf_len <= 0)) {
|
|
ret = OB_ERR_UNEXPECTED;
|
|
LOG_WARN("item buf is invalid", K(ret), K(i), K(tmp_buf_len));
|
|
} else if (FALSE_IT(clac_crc = static_cast<uint32_t>(ob_crc64(tmp_buf, tmp_buf_len)))) {
|
|
} else if (OB_UNLIKELY(block_items_map.find(clac_crc) == block_items_map.end())) {
|
|
ret = OB_ERR_UNEXPECTED;
|
|
LOG_WARN("item buf is invalid", K(ret), K(i), KP(tmp_buf), K(tmp_buf_len), K(clac_crc));
|
|
} else if (FALSE_IT(item_idx = block_items_map[clac_crc])) {
|
|
} else if (OB_UNLIKELY(item_arr[item_idx].buf_len_ != tmp_buf_len)) {
|
|
ret = OB_ERR_UNEXPECTED;
|
|
LOG_WARN("item buf len is unexpected",
|
|
K(ret), K(i), K(item_arr[item_idx]), KP(tmp_buf), K(tmp_buf_len), K(clac_crc));
|
|
}
|
|
}
|
|
|
|
if (OB_SUCC(ret)) {
|
|
ret = reader.get_next_item(tmp_buf, tmp_buf_len);
|
|
if (OB_UNLIKELY(ret != OB_ITER_END)) {
|
|
LOG_WARN("read after last item, should return OB_ITER_END", K(ret));
|
|
} else {
|
|
ret = OB_SUCCESS;
|
|
}
|
|
}
|
|
return ret;
|
|
}
|
|
|
|
int check_rw(ObSSPhysicalBlockManager &phy_blk_mgr, const std::vector<int64_t> &item_buf_len_arr,
|
|
const common::ObCompressorType compressor_type = ObCompressorType::INVALID_COMPRESSOR)
|
|
{
|
|
int ret = OB_SUCCESS;
|
|
if (OB_UNLIKELY(item_buf_len_arr.empty())) {
|
|
ret = OB_INVALID_ARGUMENT;
|
|
LOG_WARN("item_buf_len_arr is empty", K(ret));
|
|
} else {
|
|
ObArray<PhyBlockId> block_id_list;
|
|
const int64_t item_arr_size = item_buf_len_arr.size();
|
|
std::vector<ItemInfo> item_arr(item_arr_size);
|
|
const uint64_t tenant_id = MTL_ID();
|
|
ObSSLinkedPhyBlockItemWriter writer;
|
|
ObSSLinkedPhyBlockItemReader reader;
|
|
PhyBlockId entry_block_id = EMPTY_PHY_BLOCK_ID;
|
|
|
|
if (OB_FAIL(build_items(item_arr, item_buf_len_arr))) {
|
|
LOG_WARN("fail to build_items", K(ret), K(item_arr_size));
|
|
} else if (OB_FAIL(writer.init(tenant_id, phy_blk_mgr, ObSSPhyBlockType::SS_MICRO_META_CKPT_BLK, compressor_type))) {
|
|
LOG_WARN("fail to init writer", K(ret));
|
|
} else if (OB_FAIL(write_items(writer, item_arr))) {
|
|
LOG_WARN("fail to write_items", K(ret));
|
|
} else if (OB_FAIL(writer.get_entry_block(entry_block_id))) {
|
|
LOG_WARN("fail to get_entry_block", K(ret));
|
|
} else if (OB_FAIL(reader.init(entry_block_id, tenant_id, phy_blk_mgr))) {
|
|
LOG_WARN("fail to init reader", K(ret));
|
|
} else if (OB_FAIL(read_items(reader, item_arr))) {
|
|
LOG_WARN("fail to read_items", K(ret));
|
|
} else if (OB_FAIL(writer.get_block_id_list(block_id_list))) {
|
|
LOG_WARN("fail to get_block_id_list", K(ret));
|
|
} else {
|
|
for (int64_t i = 0; i < block_id_list.count(); i++) {
|
|
if (OB_UNLIKELY(block_id_list[i] != entry_block_id - i)) {
|
|
ret = OB_ERR_UNEXPECTED;
|
|
LOG_WARN("block id unexpected", K(ret), K(i), K(entry_block_id), K(block_id_list[i]));
|
|
}
|
|
}
|
|
LOG_INFO("finish item rw check", K(item_arr_size), K(block_id_list.count()), K(block_id_list));
|
|
}
|
|
}
|
|
return ret;
|
|
}
|
|
|
|
TEST_F(TestSSLinkedPhyBlock, test_basic_rw)
|
|
{
|
|
LOG_INFO("TEST: start test_basic_rw");
|
|
ObArenaAllocator allocator;
|
|
char * full_buf = static_cast<char *>(allocator.alloc(BLOCK_SIZE));
|
|
ASSERT_NE(nullptr, full_buf);
|
|
MEMSET(full_buf, 'a', BLOCK_SIZE);
|
|
ObSSPhyBlockCommonHeader common_header;
|
|
ObSSLinkedPhyBlockHeader block_header;
|
|
const int64_t block_buf_len_for_item = BLOCK_SIZE - common_header.header_size_ - block_header.get_fixed_serialize_size();
|
|
int64_t max_item_size = SS_SEG_BUF_SIZE;
|
|
bool find_ret = false;
|
|
do {
|
|
ObSSLinkedPhyBlockItemHeader item_header;
|
|
item_header.payload_size_ = max_item_size;
|
|
if (max_item_size + item_header.get_serialize_size() <= SS_SEG_BUF_SIZE) {
|
|
find_ret = true;
|
|
} else {
|
|
--max_item_size;
|
|
}
|
|
} while (!find_ret);
|
|
LOG_INFO("result: ", K(max_item_size));
|
|
|
|
LOG_INFO("=========================================TEST ITEM NO COMPRESS=========================================");
|
|
{
|
|
LOG_INFO("=========================================One Small Item=========================================");
|
|
std::vector<int64_t> item_buf_len_arr = {1};
|
|
ASSERT_EQ(OB_SUCCESS, check_rw(phy_blk_mgr_, item_buf_len_arr));
|
|
}
|
|
{
|
|
LOG_INFO("=========================================Partial Block=========================================");
|
|
std::vector<int64_t> item_buf_len_arr = {1, 2, 4, 8, 16, 8, 4};
|
|
ASSERT_EQ(OB_SUCCESS, check_rw(phy_blk_mgr_, item_buf_len_arr));
|
|
}
|
|
{
|
|
LOG_INFO("=========================================One Item Fill Single Block=========================================");
|
|
std::vector<int64_t> item_buf_len_arr = {max_item_size};
|
|
ASSERT_EQ(OB_SUCCESS, check_rw(phy_blk_mgr_, item_buf_len_arr));
|
|
}
|
|
{
|
|
LOG_INFO("=========================================One Item Less Than Block=========================================");
|
|
std::vector<int64_t> item_buf_len_arr = {max_item_size - 1};
|
|
ASSERT_EQ(OB_SUCCESS, check_rw(phy_blk_mgr_, item_buf_len_arr));
|
|
}
|
|
{
|
|
LOG_INFO("=========================================Exact Block=========================================");
|
|
const int64_t buf_len = 377;
|
|
const int64_t total_item_size = sizeof(ObSSLinkedPhyBlockItemHeader) + 13 + buf_len;
|
|
const int64_t seg_item_num = SS_SEG_BUF_SIZE / total_item_size;
|
|
const int64_t total_seg_size = sizeof(ObSSLinkedPhyBlockSegmentHeader) + SS_SEG_BUF_SIZE;
|
|
const int64_t seg_num = block_buf_len_for_item / total_seg_size;
|
|
const int64_t item_num = seg_num * seg_item_num;
|
|
std::vector<int64_t> item_buf_len_arr(item_num, buf_len);
|
|
ASSERT_EQ(OB_SUCCESS, check_rw(phy_blk_mgr_, item_buf_len_arr));
|
|
}
|
|
{
|
|
LOG_INFO("=========================================Multiple Blocks=========================================");
|
|
std::vector<int64_t> item_buf_len_arr;
|
|
int64_t reamin_size = BLOCK_SIZE * 20;
|
|
int64_t cur_item_size = 0;
|
|
while (reamin_size > 0) {
|
|
cur_item_size = ObRandom::rand(1, max_item_size);
|
|
item_buf_len_arr.push_back(cur_item_size);
|
|
reamin_size -= cur_item_size;
|
|
}
|
|
ASSERT_EQ(OB_SUCCESS, check_rw(phy_blk_mgr_, item_buf_len_arr));
|
|
}
|
|
|
|
LOG_INFO("=========================================TEST ITEM WITH COMPRESS=========================================");
|
|
{
|
|
LOG_INFO("=========================================One Small Item=========================================");
|
|
std::vector<int64_t> item_buf_len_arr = {1};
|
|
ASSERT_EQ(OB_SUCCESS, check_rw(phy_blk_mgr_, item_buf_len_arr, ObCompressorType::SNAPPY_COMPRESSOR));
|
|
}
|
|
{
|
|
LOG_INFO("=========================================Partial Block=========================================");
|
|
std::vector<int64_t> item_buf_len_arr = {1, 2, 4, 8, 16, 8, 4};
|
|
ASSERT_EQ(OB_SUCCESS, check_rw(phy_blk_mgr_, item_buf_len_arr, ObCompressorType::LZ4_COMPRESSOR));
|
|
}
|
|
{
|
|
LOG_INFO("=========================================One Item Fill Single Block=========================================");
|
|
std::vector<int64_t> item_buf_len_arr = {max_item_size};
|
|
ASSERT_EQ(OB_SUCCESS, check_rw(phy_blk_mgr_, item_buf_len_arr, ObCompressorType::ZLIB_COMPRESSOR));
|
|
}
|
|
{
|
|
LOG_INFO("=========================================One Item Less Than Block=========================================");
|
|
std::vector<int64_t> item_buf_len_arr = {max_item_size - 1};
|
|
ASSERT_EQ(OB_SUCCESS, check_rw(phy_blk_mgr_, item_buf_len_arr, ObCompressorType::ZSTD_COMPRESSOR));
|
|
}
|
|
{
|
|
LOG_INFO("=========================================Exact Block=========================================");
|
|
const int64_t buf_len = 4377;
|
|
const int64_t total_item_size = sizeof(ObSSLinkedPhyBlockItemHeader) + 13 + buf_len;
|
|
const int64_t seg_item_num = SS_SEG_BUF_SIZE / total_item_size;
|
|
const int64_t total_seg_size = sizeof(ObSSLinkedPhyBlockSegmentHeader) + SS_SEG_BUF_SIZE;
|
|
const int64_t seg_num = block_buf_len_for_item / total_seg_size;
|
|
const int64_t item_num = seg_num * seg_item_num;
|
|
std::vector<int64_t> item_buf_len_arr(item_num, buf_len);
|
|
ASSERT_EQ(OB_SUCCESS, check_rw(phy_blk_mgr_, item_buf_len_arr, ObCompressorType::SNAPPY_COMPRESSOR));
|
|
}
|
|
{
|
|
LOG_INFO("=========================================Multiple Blocks=========================================");
|
|
std::vector<int64_t> item_buf_len_arr;
|
|
int64_t reamin_size = BLOCK_SIZE * 20;
|
|
int64_t cur_item_size = 0;
|
|
while (reamin_size > 0) {
|
|
cur_item_size = ObRandom::rand(1024, max_item_size);
|
|
item_buf_len_arr.push_back(cur_item_size);
|
|
reamin_size -= cur_item_size;
|
|
}
|
|
ASSERT_EQ(OB_SUCCESS, check_rw(phy_blk_mgr_, item_buf_len_arr, ObCompressorType::SNAPPY_COMPRESSOR));
|
|
}
|
|
|
|
allocator.clear();
|
|
}
|
|
|
|
TEST_F(TestSSLinkedPhyBlock, test_ckpt_writer)
|
|
{
|
|
LOG_INFO("TEST: start test_ckpt_writer");
|
|
const uint64_t tenant_id = MTL_ID();
|
|
ObSSPhyBlockType blk_type = ObSSPhyBlockType::SS_MICRO_META_CKPT_BLK;
|
|
ASSERT_EQ(true, phy_blk_mgr_.exist_free_block(blk_type));
|
|
// 1. check 'close()': need write_block() and write_segment()
|
|
ObCompressorType compress_type = ObCompressorType::NONE_COMPRESSOR;
|
|
ObSSLinkedPhyBlockItemWriter writer;
|
|
ASSERT_EQ(OB_SUCCESS, writer.init(tenant_id, phy_blk_mgr_, blk_type, compress_type));
|
|
ItemInfo item_info;
|
|
const int64_t item_size = 2 * 1024;
|
|
ASSERT_EQ(OB_SUCCESS, item_info.init(2 * 1024));
|
|
const int64_t seg_item_cnt = SS_SEG_BUF_SIZE / item_size;
|
|
for (int64_t i = 0; i < seg_item_cnt; ++i) {
|
|
ASSERT_EQ(OB_SUCCESS, writer.write_item(item_info.buf_, item_info.buf_len_));
|
|
}
|
|
ObSSLinkedPhyBlockSegmentHeader seg_header;
|
|
ASSERT_EQ(OB_SUCCESS, writer.build_seg_header_(seg_header));
|
|
const int64_t total_seg_len = seg_header.get_serialize_size() + SS_SEG_BUF_SIZE;
|
|
while (writer.get_remaining_size() > total_seg_len) {
|
|
for (int64_t i = 0; i < seg_item_cnt; ++i) {
|
|
ASSERT_EQ(OB_SUCCESS, writer.write_item(item_info.buf_, item_info.buf_len_));
|
|
}
|
|
}
|
|
for (int64_t i = 0; i < seg_item_cnt; ++i) {
|
|
ASSERT_EQ(OB_SUCCESS, writer.write_item(item_info.buf_, item_info.buf_len_));
|
|
}
|
|
ASSERT_EQ(OB_SUCCESS, writer.close());
|
|
}
|
|
|
|
} // namespace storage
|
|
} // namespace oceanbase
|
|
|
|
int main(int argc, char **argv)
|
|
{
|
|
system("rm -f test_ss_linked_phy_block.log*");
|
|
OB_LOGGER.set_file_name("test_ss_linked_phy_block.log", true, true);
|
|
OB_LOGGER.set_log_level("INFO");
|
|
::testing::InitGoogleTest(&argc, argv);
|
|
return RUN_ALL_TESTS();
|
|
} |