hcq/oceanbase
hcq/oceanbase
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

refine dup table new gc code in leader_switch

Browse Source
This commit is contained in:
KyrielightWei 2024-04-16 20:51:19 +00:00 committed by ob-robot
parent 6cbab0038f
commit 78578299f4
11 changed files with 1005 additions and 417 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
mittest/multi_replica/CMakeLists.txt
View File

@ -34,5 +34,6 @@ ob_unittest_multi_replica(test_ob_dup_table_basic)
ob_unittest_multi_replica(test_ob_dup_table_restart)
ob_unittest_multi_replica(test_ob_dup_table_leader_switch)
ob_unittest_multi_replica(test_ob_dup_table_tablet_gc)
ob_unittest_multi_replica(test_ob_dup_table_new_gc)
ob_unittest_multi_replica(test_mds_replay_from_ctx_table)
ob_unittest_multi_replica_longer_timeout(test_multi_transfer_tx)

62
mittest/multi_replica/env/ob_multi_replica_util.h vendored
View File

@ -259,38 +259,40 @@ namespace unittest
WRITE_SQL_BY_CONN(connection, "set ob_query_timeout = 3000000000"); \
WRITE_SQL_BY_CONN(connection, "set autocommit=0");
#define RETRY_UNTIL_TIMEOUT(condition, timeout_us, retry_interval_us) \
{ \
int64_t start_time = ObTimeUtility::fast_current_time(); \
while (OB_SUCC(ret) && !(condition)) { \
if (ObTimeUtility::fast_current_time() - start_time > timeout_us) { \
ret = OB_TIMEOUT; \
break; \
} \
SERVER_LOG(INFO, "retry one time until timeout", K(condition), K(start_time), \
K(timeout_us)); \
ob_usleep(retry_interval_us); \
} \
SERVER_LOG(INFO, "retry to wait one condition successfully", K(condition), K(start_time), \
K(timeout_us), K(ObTimeUtility::fast_current_time() - start_time)); \
#define RETRY_UNTIL_TIMEOUT(condition, timeout_us, retry_interval_us) \
{ \
ret = OB_SUCCESS; \
int64_t start_time = ObTimeUtility::fast_current_time(); \
while (OB_SUCC(ret) && !(condition)) { \
if (ObTimeUtility::fast_current_time() - start_time > timeout_us) { \
ret = OB_TIMEOUT; \
break; \
} \
SERVER_LOG(INFO, "retry one time until timeout", K(condition), K(start_time), \
K(timeout_us)); \
ob_usleep(retry_interval_us); \
} \
SERVER_LOG(INFO, "retry to wait one condition successfully", K(ret), K(condition), \
K(start_time), K(timeout_us), K(ObTimeUtility::fast_current_time() - start_time)); \
}
#define RETRY_OP_UNTIL_TIMEOUT(op, condition, timeout_us, retry_interval_us) \
{ \
int64_t start_time = ObTimeUtility::fast_current_time(); \
op; \
while (OB_SUCC(ret) && !(condition)) { \
if (ObTimeUtility::fast_current_time() - start_time > timeout_us) { \
ret = OB_TIMEOUT; \
break; \
} \
SERVER_LOG(INFO, "retry opertion until timeout", K(condition), K(start_time), \
K(timeout_us)); \
ob_usleep(retry_interval_us); \
op; \
} \
SERVER_LOG(INFO, "retry to opertion successfully", K(condition), K(start_time), K(timeout_us), \
K(ObTimeUtility::fast_current_time() - start_time)); \
#define RETRY_OP_UNTIL_TIMEOUT(op, condition, timeout_us, retry_interval_us) \
{ \
ret = OB_SUCCESS; \
int64_t start_time = ObTimeUtility::fast_current_time(); \
op; \
while (OB_SUCC(ret) && !(condition)) { \
if (ObTimeUtility::fast_current_time() - start_time > timeout_us) { \
ret = OB_TIMEOUT; \
break; \
} \
SERVER_LOG(INFO, "retry opertion until timeout", K(condition), K(start_time), \
K(timeout_us)); \
ob_usleep(retry_interval_us); \
op; \
} \
SERVER_LOG(INFO, "retry to opertion successfully", K(ret), K(condition), K(start_time), \
K(timeout_us), K(ObTimeUtility::fast_current_time() - start_time)); \
}
#define WAIT_START_SERVICE_SUCCC(timeout_us, retry_interval_us) \

450
mittest/multi_replica/test_ob_dup_table_new_gc.cpp Normal file
View File

@ -0,0 +1,450 @@
/*
* Copyright (c) 2021 OceanBase
* OceanBase CE is licensed under Mulan PubL v2.
* You can use this software according to the terms and conditions of the Mulan PubL v2.
* You may obtain a copy of Mulan PubL v2 at:
* http://license.coscl.org.cn/MulanPubL-2.0
* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
* EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
* See the Mulan PubL v2 for more details.
*/
#include <gtest/gtest.h>
#define USING_LOG_PREFIX SERVER
#define protected public
#define private public
#include "env/ob_fast_bootstrap.h"
#include "env/ob_multi_replica_util.h"
#include "lib/mysqlclient/ob_mysql_result.h"
#include "storage/tx/ob_dup_table_lease.h"
#include "storage/tx/ob_dup_table_util.h"
using namespace oceanbase::transaction;
using namespace oceanbase::storage;
#define CUR_TEST_CASE_NAME ObDupTableNewGCTest
DEFINE_MULTI_ZONE_TEST_CASE_CLASS
MULTI_REPLICA_TEST_MAIN_FUNCTION(test_dup_table_new_gc_);
static const int64_t MAX_DUP_TABLE_COUNT = 10;
static bool STOP_PREPARE_SERIALIZE_DUP_TABLET = false;
namespace oceanbase
{
namespace transaction
{
OB_NOINLINE int ObLSDupTabletsMgr::process_prepare_ser_err_test_()
{
int ret = OB_SUCCESS;
if (STOP_PREPARE_SERIALIZE_DUP_TABLET) {
ret = OB_EAGAIN;
}
if (OB_FAIL(ret)) {
DUP_TABLE_LOG(INFO, "errsim prepare serialize err test in mittest", K(ret), K(ls_id_));
}
return ret;
}
} // namespace transaction
namespace unittest
{
struct DupTableBasicArg
{
uint64_t tenant_id_;
int64_t ls_id_num_;
ObSEArray<int64_t, 10> table_id_;
int64_t tablet_count_;
ObSEArray<int64_t, 100> tablet_id_array_;
DupTableBasicArg() { reset(); }
void reset()
{
tenant_id_ = 0;
ls_id_num_ = 0;
table_id_.reset();
tablet_count_ = 0;
tablet_id_array_.reset();
}
TO_STRING_KV(K(tenant_id_), K(ls_id_num_), K(table_id_), K(tablet_count_), K(tablet_id_array_));
OB_UNIS_VERSION(1);
};
OB_SERIALIZE_MEMBER(DupTableBasicArg,
tenant_id_,
ls_id_num_,
table_id_,
tablet_count_,
tablet_id_array_);
static DupTableBasicArg dup_basic_arg_;
static const int64_t CUR_GC_INTERVAL = 2 * 1000 * 1000;
void reduce_dup_table_gc_interval(int64_t tenant_id, int64_t cur_gc_interval)
{
share::ObTenantSwitchGuard tenant_guard;
ASSERT_EQ(OB_SUCCESS, tenant_guard.switch_to(tenant_id));
MTL(transaction::ObTransService *)->dup_tablet_scan_task_.scan_task_execute_interval_ =
cur_gc_interval;
MTL(transaction::ObTransService *)->dup_tablet_scan_task_.max_execute_interval_ = cur_gc_interval;
oceanbase::transaction::ObLSDupTabletsMgr::MAX_READABLE_SET_SER_INTERVAL = CUR_GC_INTERVAL;
}
TEST_F(GET_ZONE_TEST_CLASS_NAME(1), create_dup_table)
{
int ret = OB_SUCCESS;
CREATE_TEST_TENANT(test_tenant_id);
SERVER_LOG(INFO, "[ObMultiReplicaTestBase] create test tenant success", K(test_tenant_id));
reduce_dup_table_gc_interval(test_tenant_id, CUR_GC_INTERVAL);
common::ObMySQLProxy &test_tenant_sql_proxy = get_curr_simple_server().get_sql_proxy2();
ACQUIRE_CONN_FROM_SQL_PROXY(test_conn, test_tenant_sql_proxy);
ACQUIRE_CONN_FROM_SQL_PROXY(sys_conn, get_curr_simple_server().get_sql_proxy());
std::string primary_zone_sql = "ALTER TENANT " + std::string(DEFAULT_TEST_TENANT_NAME)
+ " set primary_zone='zone1; zone3; zone2';";
WRITE_SQL_BY_CONN(test_conn, primary_zone_sql.c_str());
for (int i = 0; i < MAX_DUP_TABLE_COUNT; i++) {
std::string tname = "test_t" + std::to_string(i);
std::string create_table_sql =
"CREATE TABLE " + tname
+ "( "
"id_x int, "
"id_y int, "
"id_z int, "
"PRIMARY KEY(id_x)"
") duplicate_scope='cluster' PARTITION BY hash(id_x) partitions 10;";
std::string get_table_id_sql = "select table_id, duplicate_scope from "
"oceanbase.__all_table where table_name = '"
+ tname + "' ";
WRITE_SQL_BY_CONN(test_conn, create_table_sql.c_str());
READ_SQL_BY_CONN(test_conn, table_info_result, get_table_id_sql.c_str());
ASSERT_EQ(OB_SUCCESS, table_info_result->next());
int64_t table_id = 0;
int64_t dup_scope = 0;
ASSERT_EQ(OB_SUCCESS, table_info_result->get_int("table_id", table_id));
ASSERT_EQ(OB_SUCCESS, table_info_result->get_int("duplicate_scope", dup_scope));
ASSERT_EQ(true, table_id > 0);
ASSERT_EQ(true, dup_scope != 0);
ASSERT_EQ(OB_SUCCESS, dup_basic_arg_.table_id_.push_back(table_id));
std::string tablet_count_sql =
"select count(*), ls_id from oceanbase.__all_tablet_to_ls where table_id = "
+ std::to_string(table_id) + " group by ls_id order by count(*)";
READ_SQL_BY_CONN(test_conn, tablet_count_result, tablet_count_sql.c_str());
int64_t tablet_count = 0;
int64_t ls_id_num = 0;
ASSERT_EQ(OB_SUCCESS, tablet_count_result->next());
ASSERT_EQ(OB_SUCCESS, tablet_count_result->get_int("count(*)", tablet_count));
ASSERT_EQ(OB_SUCCESS, tablet_count_result->get_int("ls_id", ls_id_num));
ASSERT_EQ(10, tablet_count);
ASSERT_EQ(true, share::ObLSID(ls_id_num).is_valid());
if (dup_basic_arg_.ls_id_num_ <= 0) {
dup_basic_arg_.ls_id_num_ = ls_id_num;
} else {
ASSERT_EQ(dup_basic_arg_.ls_id_num_, ls_id_num);
}
dup_basic_arg_.tablet_count_ += tablet_count;
std::string tablet_id_sql =
"select tablet_id from oceanbase.__all_tablet_to_ls where table_id = "
+ std::to_string(table_id) + " and ls_id = " + std::to_string(ls_id_num);
READ_SQL_BY_CONN(test_conn, tablet_id_reult, tablet_id_sql.c_str());
while (OB_SUCC(tablet_id_reult->next())) {
int64_t id = 0;
ASSERT_EQ(OB_SUCCESS, tablet_id_reult->get_int("tablet_id", id));
ASSERT_EQ(true, ObTabletID(id).is_valid());
ASSERT_EQ(OB_SUCCESS, dup_basic_arg_.tablet_id_array_.push_back(id));
}
}
dup_basic_arg_.tenant_id_ = test_tenant_id;
std::string tmp_str;
ASSERT_EQ(OB_SUCCESS, EventArgSerTool<DupTableBasicArg>::serialize_arg(dup_basic_arg_, tmp_str));
ASSERT_EQ(OB_SUCCESS, finish_event("CREATE_DUP_TABLE", tmp_str));
}
TEST_F(GET_ZONE_TEST_CLASS_NAME(2), leader_switch_without_gc)
{
int ret = OB_SUCCESS;
std::string tmp_event_val;
ASSERT_EQ(OB_SUCCESS, wait_event_finish("CREATE_DUP_TABLE", tmp_event_val, 30 * 60 * 1000));
ASSERT_EQ(OB_SUCCESS,
EventArgSerTool<DupTableBasicArg>::deserialize_arg(dup_basic_arg_, tmp_event_val));
reduce_dup_table_gc_interval(dup_basic_arg_.tenant_id_, CUR_GC_INTERVAL);
ASSERT_EQ(OB_SUCCESS, get_curr_simple_server().init_sql_proxy2());
common::ObMySQLProxy &sys_sql_proxy = get_curr_simple_server().get_sql_proxy();
ACQUIRE_CONN_FROM_SQL_PROXY(sys_conn, sys_sql_proxy);
GET_LS(dup_basic_arg_.tenant_id_, dup_basic_arg_.ls_id_num_, ls_handle);
RETRY_UNTIL_TIMEOUT(
ls_handle.get_ls()->dup_table_ls_handler_.is_inited()
&& ls_handle.get_ls()->dup_table_ls_handler_.tablets_mgr_ptr_->get_readable_tablet_count()
== dup_basic_arg_.tablet_count_,
30 * 1000 * 1000, 1 * 100 * 1000);
ASSERT_EQ(ret, OB_SUCCESS);
RETRY_UNTIL_TIMEOUT(!ls_handle.get_ls()->dup_table_ls_handler_.is_master(), 20 * 1000 * 1000,
100 * 1000);
ASSERT_EQ(ret, OB_SUCCESS);
usleep(ObDupTabletScanTask::DUP_TABLET_SCAN_INTERVAL + 4 * CUR_GC_INTERVAL);
std::string ls_id_str = std::to_string(dup_basic_arg_.ls_id_num_);
std::string zone2_ip = local_ip_ + ":" + std::to_string(rpc_ports_[1]);
std::string switch_leader_sql = "alter system switch replica leader ls=" + ls_id_str + " server='"
+ zone2_ip + "' tenant='tt1';";
WRITE_SQL_BY_CONN(sys_conn, switch_leader_sql.c_str());
RETRY_OP_UNTIL_TIMEOUT(
ASSERT_EQ(
ls_handle.get_ls()->dup_table_ls_handler_.tablets_mgr_ptr_->get_readable_tablet_count(),
dup_basic_arg_.tablet_count_),
ls_handle.get_ls()->dup_table_ls_handler_.is_master(), 20 * 1000 * 1000, 50 * 1000);
ASSERT_EQ(ret, OB_SUCCESS);
RETRY_UNTIL_TIMEOUT(
ls_handle.get_ls()->dup_table_ls_handler_.tablets_mgr_ptr_->get_readable_tablet_count()
< dup_basic_arg_.tablet_count_,
10 * 1000 * 1000, 50 * 1000);
EXPECT_EQ(ret, OB_TIMEOUT);
RETRY_UNTIL_TIMEOUT(
ls_handle.get_ls()->dup_table_ls_handler_.tablets_mgr_ptr_->get_readable_tablet_count()
== dup_basic_arg_.tablet_count_,
20 * 1000 * 1000, 100 * 1000);
ASSERT_EQ(ret, OB_SUCCESS);
std::string zone1_ip = local_ip_ + ":" + std::to_string(rpc_ports_[0]);
std::string switch_leader_to_zone1_sql = "alter system switch replica leader ls=" + ls_id_str
+ " server='" + zone1_ip + "' tenant='tt1';";
WRITE_SQL_BY_CONN(sys_conn, switch_leader_to_zone1_sql.c_str());
RETRY_UNTIL_TIMEOUT(!ls_handle.get_ls()->dup_table_ls_handler_.is_master(), 20 * 1000 * 1000,
100 * 1000);
ASSERT_EQ(ret, OB_SUCCESS);
ASSERT_EQ(OB_SUCCESS, finish_event("SWITCH_LEADER_TO_ZONE2_WITHOUT_GC", ""));
}
TEST_F(GET_ZONE_TEST_CLASS_NAME(1), leader_revoke_in_gc)
{
int ret = OB_SUCCESS;
std::string tmp_event_val;
ASSERT_EQ(OB_SUCCESS,
wait_event_finish("SWITCH_LEADER_TO_ZONE2_WITHOUT_GC", tmp_event_val, 30 * 60 * 1000));
common::ObMySQLProxy &test_tenant_sql_proxy = get_curr_simple_server().get_sql_proxy2();
ACQUIRE_CONN_FROM_SQL_PROXY(test_conn, test_tenant_sql_proxy);
GET_LS(dup_basic_arg_.tenant_id_, dup_basic_arg_.ls_id_num_, ls_handle);
RETRY_UNTIL_TIMEOUT(ls_handle.get_ls()->dup_table_ls_handler_.is_master(), 20 * 1000 * 1000,
100 * 1000);
ASSERT_EQ(ret, OB_SUCCESS);
RETRY_UNTIL_TIMEOUT(
ls_handle.get_ls()->dup_table_ls_handler_.tablets_mgr_ptr_->get_readable_tablet_count()
== dup_basic_arg_.tablet_count_,
20 * 1000 * 1000, 100 * 1000);
ASSERT_EQ(ret, OB_SUCCESS);
TRANS_LOG(INFO, "[ObMultiReplicaTestBase] print dup tablet stat info 1", K(ret),
KPC(ls_handle.get_ls()->dup_table_ls_handler_.tablets_mgr_ptr_));
{
std::string tname = "test_t" + std::to_string(1);
std::string delete_table_sql = "DROP TABLE " + tname;
WRITE_SQL_BY_CONN(test_conn, delete_table_sql.c_str());
std::string get_table_id_sql = "select table_id, duplicate_scope from "
"oceanbase.__all_table where table_name = '"
+ tname + "' ";
READ_SQL_BY_CONN(test_conn, table_info_result, get_table_id_sql.c_str());
ASSERT_EQ(OB_ITER_END, table_info_result->next());
}
// gc t1 successfully
RETRY_UNTIL_TIMEOUT(
ls_handle.get_ls()->dup_table_ls_handler_.tablets_mgr_ptr_->get_readable_tablet_count()
== dup_basic_arg_.tablet_count_ - 10,
20 * 1000 * 1000, 100 * 1000);
TRANS_LOG(INFO, "[ObMultiReplicaTestBase] print dup tablet stat info 1.1", K(ret),
KPC(ls_handle.get_ls()->dup_table_ls_handler_.tablets_mgr_ptr_));
ls_handle.get_ls()->dup_table_ls_handler_.tablets_mgr_ptr_->print_tablet_diag_info_log(true);
ASSERT_EQ(ret, OB_SUCCESS);
RETRY_UNTIL_TIMEOUT(
(ls_handle.get_ls()->dup_table_ls_handler_.tablets_mgr_ptr_->removing_old_set_->size() == 0),
20 * 1000 * 1000, 100 * 1000);
ASSERT_EQ(ret, OB_SUCCESS);
STOP_PREPARE_SERIALIZE_DUP_TABLET = true;
{
std::string tname = "test_t" + std::to_string(2);
std::string delete_table_sql = "DROP TABLE " + tname;
WRITE_SQL_BY_CONN(test_conn, delete_table_sql.c_str());
std::string get_table_id_sql = "select table_id, duplicate_scope from "
"oceanbase.__all_table where table_name = '"
+ tname + "' ";
READ_SQL_BY_CONN(test_conn, table_info_result, get_table_id_sql.c_str());
ASSERT_EQ(OB_ITER_END, table_info_result->next());
}
TRANS_LOG(INFO, "[ObMultiReplicaTestBase] print dup tablet stat info 2", K(ret),
KPC(ls_handle.get_ls()->dup_table_ls_handler_.tablets_mgr_ptr_));
RETRY_UNTIL_TIMEOUT(
(ls_handle.get_ls()->dup_table_ls_handler_.tablets_mgr_ptr_->removing_old_set_->size() > 0),
20 * 1000 * 1000, 100 * 1000);
ls_handle.get_ls()->dup_table_ls_handler_.tablets_mgr_ptr_->print_tablet_diag_info_log(true);
ASSERT_EQ(ret, OB_SUCCESS);
TRANS_LOG(INFO, "[ObMultiReplicaTestBase] print dup tablet stat info 3", K(ret),
KPC(ls_handle.get_ls()->dup_table_ls_handler_.tablets_mgr_ptr_));
ASSERT_EQ(ls_handle.get_ls()->dup_table_ls_handler_.tablets_mgr_ptr_->get_readable_tablet_count(),
dup_basic_arg_.tablet_count_ - 10 /*readable 90 + removing 10*/);
ASSERT_EQ(ls_handle.get_ls()->dup_table_ls_handler_.tablets_mgr_ptr_->removing_old_set_->size(),
10);
ASSERT_EQ(OB_SUCCESS, finish_event("DROP_TABLE_WITHOUT_LOG", ""));
}
TEST_F(GET_ZONE_TEST_CLASS_NAME(2), continue_gc_in_new_leader)
{
int ret = OB_SUCCESS;
std::string tmp_event_val;
ASSERT_EQ(OB_SUCCESS, wait_event_finish("DROP_TABLE_WITHOUT_LOG", tmp_event_val, 30 * 60 * 1000));
STOP_PREPARE_SERIALIZE_DUP_TABLET = true;
common::ObMySQLProxy &test_tenant_sql_proxy = get_curr_simple_server().get_sql_proxy2();
ACQUIRE_CONN_FROM_SQL_PROXY(test_conn, test_tenant_sql_proxy);
ACQUIRE_CONN_FROM_SQL_PROXY(sys_conn, get_curr_simple_server().get_sql_proxy());
GET_LS(dup_basic_arg_.tenant_id_, dup_basic_arg_.ls_id_num_, ls_handle);
std::string ls_id_str = std::to_string(dup_basic_arg_.ls_id_num_);
std::string zone2_ip = local_ip_ + ":" + std::to_string(rpc_ports_[1]);
std::string switch_leader_sql = "alter system switch replica leader ls=" + ls_id_str + " server='"
+ zone2_ip + "' tenant='tt1';";
WRITE_SQL_BY_CONN(sys_conn, switch_leader_sql.c_str());
RETRY_UNTIL_TIMEOUT(ls_handle.get_ls()->dup_table_ls_handler_.is_master(), 20 * 1000 * 1000,
100 * 1000);
ASSERT_EQ(ret, OB_SUCCESS);
RETRY_UNTIL_TIMEOUT(
ls_handle.get_ls()->dup_table_ls_handler_.tablets_mgr_ptr_->get_readable_tablet_count()
< dup_basic_arg_.tablet_count_ - 10,
20 * 1000 * 1000, 100 * 1000);
TRANS_LOG(INFO, "[ObMultiReplicaTestBase] print dup tablet stat info 4", K(ret),
KPC(ls_handle.get_ls()->dup_table_ls_handler_.tablets_mgr_ptr_));
ls_handle.get_ls()->dup_table_ls_handler_.tablets_mgr_ptr_->print_tablet_diag_info_log(true);
ASSERT_EQ(ret, OB_TIMEOUT);
RETRY_UNTIL_TIMEOUT(
ls_handle.get_ls()->dup_table_ls_handler_.tablets_mgr_ptr_->get_readable_tablet_count()
== dup_basic_arg_.tablet_count_ - 10 /*readable 90 + removing 10*/,
20 * 1000 * 1000, 100 * 1000);
ASSERT_EQ(ret, OB_SUCCESS);
ASSERT_EQ(ls_handle.get_ls()->dup_table_ls_handler_.tablets_mgr_ptr_->removing_old_set_->size(),
10);
STOP_PREPARE_SERIALIZE_DUP_TABLET = false;
RETRY_UNTIL_TIMEOUT(
ls_handle.get_ls()->dup_table_ls_handler_.tablets_mgr_ptr_->get_readable_tablet_count()
== dup_basic_arg_.tablet_count_ - 10 * 2,
20 * 1000 * 1000, 100 * 1000);
ASSERT_EQ(ret, OB_SUCCESS);
ASSERT_EQ(OB_SUCCESS, finish_event("CONTINUE_GC_IN_ZONE2", ""));
}
TEST_F(GET_ZONE_TEST_CLASS_NAME(1), check_gc_result_in_old_leader)
{
int ret = OB_SUCCESS;
std::string tmp_event_val;
ASSERT_EQ(OB_SUCCESS, wait_event_finish("CONTINUE_GC_IN_ZONE2", tmp_event_val, 30 * 60 * 1000));
common::ObMySQLProxy &test_tenant_sql_proxy = get_curr_simple_server().get_sql_proxy2();
ACQUIRE_CONN_FROM_SQL_PROXY(test_conn, test_tenant_sql_proxy);
ACQUIRE_CONN_FROM_SQL_PROXY(sys_conn, get_curr_simple_server().get_sql_proxy());
GET_LS(dup_basic_arg_.tenant_id_, dup_basic_arg_.ls_id_num_, ls_handle);
RETRY_UNTIL_TIMEOUT(!ls_handle.get_ls()->dup_table_ls_handler_.is_master(), 20 * 1000 * 1000,
100 * 1000);
ASSERT_EQ(ret, OB_SUCCESS);
RETRY_UNTIL_TIMEOUT(
ls_handle.get_ls()->dup_table_ls_handler_.tablets_mgr_ptr_->get_readable_tablet_count()
== dup_basic_arg_.tablet_count_ - 10 * 2,
20 * 1000 * 1000, 100 * 1000);
ASSERT_EQ(ret, OB_SUCCESS);
std::string ls_id_str = std::to_string(dup_basic_arg_.ls_id_num_);
std::string zone1_ip = local_ip_ + ":" + std::to_string(rpc_ports_[0]);
std::string switch_leader_to_zone1_sql = "alter system switch replica leader ls=" + ls_id_str
+ " server='" + zone1_ip + "' tenant='tt1';";
WRITE_SQL_BY_CONN(sys_conn, switch_leader_to_zone1_sql.c_str());
RETRY_UNTIL_TIMEOUT(ls_handle.get_ls()->dup_table_ls_handler_.is_master(), 20 * 1000 * 1000,
100 * 1000);
ASSERT_EQ(ret, OB_SUCCESS);
RETRY_UNTIL_TIMEOUT(
ls_handle.get_ls()->dup_table_ls_handler_.tablets_mgr_ptr_->get_readable_tablet_count()
== dup_basic_arg_.tablet_count_ - 10 * 2,
20 * 1000 * 1000, 100 * 1000);
ASSERT_EQ(ret, OB_SUCCESS);
}
// TEST_F(GET_ZONE_TEST_CLASS_NAME(1), random_error)
// {
// ASSERT_EQ(ObTimeUtility::current_time() % 4 != 1, true);
// }
//
// TEST_F(GET_ZONE_TEST_CLASS_NAME(2), random_error)
// {
// ASSERT_EQ(ObTimeUtility::current_time() % 4 != 2, true);
// }
//
// TEST_F(GET_ZONE_TEST_CLASS_NAME(3), random_error)
// {
// ASSERT_EQ(ObTimeUtility::current_time() % 4 != 3, true);
// }
} // namespace unittest
} // namespace oceanbase

4
src/storage/tx/ob_dup_table_base.cpp
View File

@ -560,10 +560,8 @@ int ObDupTableLogOperator::submit_log_entry()
int64_t max_ser_size = 0;
bool submit_result = false;
DupLogTypeArray type_array;
if (OB_SUCC(ret)) {
if (OB_TMP_FAIL(tablet_mgr_ptr_->scan_readable_set_for_gc())) {
DUP_TABLE_LOG(WARN, "scan readable set failed", K(tmp_ret));
}
if (OB_FAIL(prepare_serialize_log_entry_(max_ser_size, type_array))) {
DUP_TABLE_LOG(WARN, "prepare serialize log entry failed", K(ret));
} else if (!type_array.empty()) {

7
src/storage/tx/ob_dup_table_base.h
View File

@ -82,6 +82,8 @@ struct DupTableInterfaceStat
int64_t dup_table_lease_log_sync_total_time_;
int64_t dup_table_tablet_log_sync_total_time_;
int64_t dup_table_ls_leader_takeover_ts_;
void reset()
{
dup_table_follower_read_succ_cnt_ = 0;
@ -101,6 +103,8 @@ struct DupTableInterfaceStat
dup_table_log_deser_total_time_ = 0;
dup_table_lease_log_sync_total_time_ = 0;
dup_table_tablet_log_sync_total_time_ = 0;
dup_table_ls_leader_takeover_ts_ = 0;
}
TO_STRING_KV(K(dup_table_follower_read_succ_cnt_),
@ -115,7 +119,8 @@ struct DupTableInterfaceStat
K(dup_table_log_replay_total_time_),
K(dup_table_log_deser_total_time_),
K(dup_table_lease_log_sync_total_time_),
K(dup_table_tablet_log_sync_total_time_));
K(dup_table_tablet_log_sync_total_time_),
K(dup_table_ls_leader_takeover_ts_));
};
#define DUP_LEASE_LIFE_PREFIX "[DupLeaseLife] "

524
src/storage/tx/ob_dup_table_tablets.cpp
View File

@ -9,6 +9,7 @@
// See the Mulan PubL v2 for more details.
#include "lib/utility/ob_tracepoint.h"
#include "storage/tx/ob_trans_service.h"
#include "ob_dup_table_base.h"
#include "ob_dup_table_tablets.h"
#include "ob_dup_table_util.h"
@ -28,6 +29,9 @@ namespace transaction
int64_t ObLSDupTabletsMgr::GC_DUP_TABLETS_TIME_INTERVAL = 5 * 60 * 1000 * 1000L; // 5 min
int64_t ObLSDupTabletsMgr::GC_DUP_TABLETS_FAILED_TIMEOUT =
5 * GC_DUP_TABLETS_TIME_INTERVAL; // 25 min
int64_t ObLSDupTabletsMgr::MAX_READABLE_SET_SER_INTERVAL = 30 * 1000 * 1000L; //30s
const int64_t ObLSDupTabletsMgr::GC_TIMEOUT = 1 * 1000 * 1000L; // 1s
const int64_t ObLSDupTabletsMgr::RESERVED_FREE_SET_COUNT = 64;
@ -444,10 +448,8 @@ void ObLSDupTabletsMgr::reset()
ls_id_.reset();
ATOMIC_STORE(&is_stopped_, true);
ATOMIC_STORE(&is_master_, false);
last_gc_succ_time_ = 0;
last_no_free_set_time_ = 0;
extra_free_set_alloc_count_ = 0;
tablet_gc_window_ = 2 * ObDupTabletScanTask::DUP_TABLET_SCAN_INTERVAL;
gc_start_time_ = 0;
readable_set_in_gc_ = nullptr;
@ -699,166 +701,177 @@ ERRSIM_POINT_DEF(ERRSIM_DUP_TABLE_GC_RIGHT_NOW);
int ObLSDupTabletsMgr::gc_tmporary_dup_tablets(const int64_t gc_ts, const int64_t max_task_interval)
{
int ret = OB_SUCCESS;
SpinWLockGuard guard(dup_tablets_lock_);
int gc_tablet_cnt = 0;
ObTabletID tmp_id;
// run gc now
if (OB_FAIL(ERRSIM_DUP_TABLE_GC_RIGHT_NOW)) {
ret = OB_SUCCESS;
last_gc_succ_time_ = gc_ts - GC_DUP_TABLETS_TIME_INTERVAL;
DUP_TABLE_LOG(WARN, "use errsim to invoke gc", KR(ret), K(last_gc_succ_time_), K(gc_ts),
K(max_task_interval));
}
if (0 > (gc_ts - last_gc_succ_time_) || 0 > last_gc_succ_time_ || 0 > gc_ts) {
ret = OB_INVALID_ARGUMENT;
DUP_TABLE_LOG(WARN, "Invalid gc_ts or last_gc_time_", KR(ret), K(last_gc_succ_time_), K(gc_ts));
} else if ((gc_ts - last_gc_succ_time_) < GC_DUP_TABLETS_TIME_INTERVAL) {
DUP_TABLE_LOG(DEBUG, "not need gc now", K(last_gc_succ_time_));
} else {
tablet_gc_window_ = 2
* (max_task_interval > ObDupTabletScanTask::DUP_TABLET_SCAN_INTERVAL
? max_task_interval
: ObDupTabletScanTask::DUP_TABLET_SCAN_INTERVAL);
int64_t gc_timeout = 0;
if ((gc_ts - last_gc_succ_time_) > GC_DUP_TABLETS_FAILED_TIMEOUT && last_gc_succ_time_ != 0) {
gc_timeout = INT64_MAX;
DUP_TABLE_LOG(WARN,
"gc failed too much times, this time should not break",
K(ret),
K(gc_timeout),
K(GC_DUP_TABLETS_FAILED_TIMEOUT),
K(gc_ts),
K(last_gc_succ_time_));
} else {
gc_timeout = GC_TIMEOUT;
}
int64_t gc_start_time = ObTimeUtility::fast_current_time();
/**
* Gc readable tablet set
* */
DupTabletChangeMap *old_tablet_set = nullptr;
DupTabletSetCommonHeader old_tablet_common_header;
old_tablet_common_header.set_old();
old_tablet_common_header.set_invalid_unique_id();
if (OB_FAIL(ret)) {
} else if (OB_FAIL(get_target_tablet_set_(old_tablet_common_header, old_tablet_set))) {
DUP_TABLE_LOG(WARN, "get old tablet set failed, need skip gc readable tablets", K(ret),
KPC(old_tablet_set));
ret = OB_SUCCESS;
} else if (!old_tablet_set->get_change_status()->is_modifiable()) {
ret = OB_EAGAIN; // should not update gc succ time to increase gc freq
DUP_TABLE_LOG(INFO, "old tablet set can not be modified, skip gc readable tablets", K(ret),
KPC(old_tablet_set));
} else if (old_tablet_set->get_related_set_op_type() == DupTableRelatedSetOpType::OLD_GC) {
DLIST_FOREACH(readable_tablets_ptr, readable_tablets_list_)
{
if (readable_tablets_ptr->is_logging()) {
// do nothing
} else {
GcDiscardedDupTabletHandler readable_gc_handler(
gc_ts, tablet_gc_window_, readable_tablets_ptr->get_common_header(), *old_tablet_set);
if (OB_FAIL(hash_for_each_remove_with_timeout(tmp_id, *readable_tablets_ptr,
readable_gc_handler, gc_timeout))) {
DUP_TABLE_LOG(WARN, "remove readable tablets failed", KR(ret),
K(readable_gc_handler.get_gc_tablet_cnt()));
} else if (OB_FAIL(readable_gc_handler.get_ret())) {
// if fail, not update last gc succ time to increase gc freqency
DUP_TABLE_LOG(WARN, "remove readable tablets failed, may need retry", KR(ret),
K(readable_gc_handler.get_gc_tablet_cnt()));
}
gc_tablet_cnt += readable_gc_handler.get_gc_tablet_cnt();
}
}
} else {
DUP_TABLE_LOG(WARN, "related set type not match", K(ret), KPC(removing_old_set_));
}
/**
* Gc new tablet set
* */
DupTabletChangeMap *changing_new_set = nullptr;
DupTabletSetCommonHeader new_tablet_common_header;
new_tablet_common_header.set_new();
new_tablet_common_header.set_invalid_unique_id();
if (OB_FAIL(ret)) {
// do nothing
} else if (OB_FAIL(get_target_tablet_set_(new_tablet_common_header, changing_new_set))) {
DUP_TABLE_LOG(WARN, "get changing new set failed", K(ret), KPC(changing_new_set));
} else if (OB_NOT_NULL(changing_new_set)) {
if (changing_new_set->empty()) {
// do nothing
DUP_TABLE_LOG(DEBUG, "changing_new_set is empty, not need gc", K(ret));
} else {
GcDiscardedDupTabletHandler new_gc_handler(
gc_ts, tablet_gc_window_, changing_new_set->get_common_header(), *old_tablet_set);
if (OB_FAIL(hash_for_each_remove_with_timeout(tmp_id, *changing_new_set, new_gc_handler,
gc_timeout))) {
DUP_TABLE_LOG(WARN, "remove new tablets failed", KR(ret));
}
// collect gc in new tablets count
gc_tablet_cnt += new_gc_handler.get_gc_tablet_cnt();
}
}
// collect gc readable tablet
if (OB_SUCC(ret)) {
last_gc_succ_time_ = gc_ts;
} else if (OB_TIMEOUT == ret) {
DUP_TABLE_LOG(WARN, "gc tablets failed, scan all tablets set cost too much time", K(ret),
K(gc_start_time), K(gc_timeout), K(gc_tablet_cnt));
} else if (OB_EAGAIN == ret) {
ret = OB_SUCCESS;
}
if (0 != gc_tablet_cnt) {
DUP_TABLE_LOG(INFO, "finish gc dup tablet on time", K(ret), KPC(changing_new_set_),
KPC(removing_old_set_), K(readable_tablets_list_.get_size()), K(gc_tablet_cnt));
}
}
// SpinWLockGuard guard(dup_tablets_lock_);
// int gc_tablet_cnt = 0;
// ObTabletID tmp_id;
//
// // run gc now
// if (OB_FAIL(ERRSIM_DUP_TABLE_GC_RIGHT_NOW)) {
// ret = OB_SUCCESS;
// last_gc_succ_time_ = gc_ts - GC_DUP_TABLETS_TIME_INTERVAL;
// DUP_TABLE_LOG(WARN, "use errsim to invoke gc", KR(ret), K(last_gc_succ_time_), K(gc_ts),
// K(max_task_interval));
// }
//
// if (0 >= (gc_ts - last_gc_succ_time_) || 0 >= last_gc_succ_time_ || 0 >= gc_ts) {
// ret = OB_INVALID_ARGUMENT;
// DUP_TABLE_LOG(WARN, "Invalid gc_ts or last_gc_time_", KR(ret), K(last_gc_succ_time_), K(gc_ts));
// } else if ((gc_ts - last_gc_succ_time_) < GC_DUP_TABLETS_TIME_INTERVAL) {
// DUP_TABLE_LOG(DEBUG, "not need gc now", K(last_gc_succ_time_));
// } else {
// tablet_gc_window_ = 2
// * (max_task_interval > ObDupTabletScanTask::DUP_TABLET_SCAN_INTERVAL
// ? max_task_interval
// : ObDupTabletScanTask::DUP_TABLET_SCAN_INTERVAL);
//
// int64_t gc_timeout = 0;
// if ((gc_ts - last_gc_succ_time_) > GC_DUP_TABLETS_FAILED_TIMEOUT && last_gc_succ_time_ != 0) {
// gc_timeout = INT64_MAX;
// DUP_TABLE_LOG(WARN,
// "gc failed too much times, this time should not break",
// K(ret),
// K(gc_timeout),
// K(GC_DUP_TABLETS_FAILED_TIMEOUT),
// K(gc_ts),
// K(last_gc_succ_time_));
// } else {
// gc_timeout = GC_TIMEOUT;
// }
//
// int64_t gc_start_time = ObTimeUtility::fast_current_time();
//
// /**
// * Gc readable tablet set
// * */
// DupTabletChangeMap *old_tablet_set = nullptr;
// DupTabletSetCommonHeader old_tablet_common_header;
// old_tablet_common_header.set_old();
// old_tablet_common_header.set_invalid_unique_id();
//
// if (OB_FAIL(ret)) {
// } else if (OB_FAIL(get_target_tablet_set_(old_tablet_common_header, old_tablet_set))) {
// DUP_TABLE_LOG(WARN, "get old tablet set failed, need skip gc readable tablets", K(ret),
// KPC(old_tablet_set));
// ret = OB_SUCCESS;
// } else if (!old_tablet_set->get_change_status()->is_modifiable()) {
// ret = OB_EAGAIN; // should not update gc succ time to increase gc freq
// DUP_TABLE_LOG(INFO, "old tablet set can not be modified, skip gc readable tablets", K(ret),
// KPC(old_tablet_set));
// } else if (old_tablet_set->get_related_set_op_type() == DupTableRelatedSetOpType::OLD_GC) {
// DLIST_FOREACH(readable_tablets_ptr, readable_tablets_list_)
// {
// if (readable_tablets_ptr->is_logging()) {
// // do nothing
// } else {
// GcDiscardedDupTabletHandler readable_gc_handler(
// gc_ts, tablet_gc_window_, readable_tablets_ptr->get_common_header(), *old_tablet_set);
// if (OB_FAIL(hash_for_each_remove_with_timeout(tmp_id, *readable_tablets_ptr,
// readable_gc_handler, gc_timeout))) {
// DUP_TABLE_LOG(WARN, "remove readable tablets failed", KR(ret),
// K(readable_gc_handler.get_gc_tablet_cnt()));
// } else if (OB_FAIL(readable_gc_handler.get_ret())) {
// // if fail, not update last gc succ time to increase gc freqency
// DUP_TABLE_LOG(WARN, "remove readable tablets failed, may need retry", KR(ret),
// K(readable_gc_handler.get_gc_tablet_cnt()));
// }
// gc_tablet_cnt += readable_gc_handler.get_gc_tablet_cnt();
// }
// }
// } else {
// DUP_TABLE_LOG(WARN, "related set type not match", K(ret), KPC(removing_old_set_));
// }
//
// /**
// * Gc new tablet set
// * */
// DupTabletChangeMap *changing_new_set = nullptr;
// DupTabletSetCommonHeader new_tablet_common_header;
// new_tablet_common_header.set_new();
// new_tablet_common_header.set_invalid_unique_id();
// if (OB_FAIL(ret)) {
// // do nothing
// } else if (OB_FAIL(get_target_tablet_set_(new_tablet_common_header, changing_new_set))) {
// DUP_TABLE_LOG(WARN, "get changing new set failed", K(ret), KPC(changing_new_set));
// } else if (OB_NOT_NULL(changing_new_set)) {
// if (changing_new_set->empty()) {
// // do nothing
// DUP_TABLE_LOG(DEBUG, "changing_new_set is empty, not need gc", K(ret));
// } else {
// GcDiscardedDupTabletHandler new_gc_handler(
// gc_ts, tablet_gc_window_, changing_new_set->get_common_header(), *old_tablet_set);
//
// if (OB_FAIL(hash_for_each_remove_with_timeout(tmp_id, *changing_new_set, new_gc_handler,
// gc_timeout))) {
// DUP_TABLE_LOG(WARN, "remove new tablets failed", KR(ret));
// }
// // collect gc in new tablets count
// gc_tablet_cnt += new_gc_handler.get_gc_tablet_cnt();
// }
// }
// // collect gc readable tablet
// if (OB_SUCC(ret)) {
// last_gc_succ_time_ = gc_ts;
// } else if (OB_TIMEOUT == ret) {
// DUP_TABLE_LOG(WARN, "gc tablets failed, scan all tablets set cost too much time", K(ret),
// K(gc_start_time), K(gc_timeout), K(gc_tablet_cnt));
// } else if (OB_EAGAIN == ret) {
// ret = OB_SUCCESS;
// }
//
// if (0 != gc_tablet_cnt) {
// DUP_TABLE_LOG(INFO, "finish gc dup tablet on time", K(ret), KPC(changing_new_set_),
// KPC(removing_old_set_), K(readable_tablets_list_.get_size()), K(gc_tablet_cnt));
// }
// }
return ret;
}
DupTabletChangeMap *ObLSDupTabletsMgr::get_need_gc_set_(bool &new_round)
DupTabletChangeMap *ObLSDupTabletsMgr::get_need_gc_set_()
{
if (OB_ISNULL(readable_set_in_gc_) || gc_start_time_ <= 0) {
gc_start_time_ = ObTimeUtility::fast_current_time();
if (!readable_tablets_list_.is_empty()) {
readable_set_in_gc_ = readable_tablets_list_.get_first();
}
}
if (OB_NOT_NULL(readable_set_in_gc_)) {
if (!readable_set_in_gc_->get_common_header().is_readable_set()) {
DUP_TABLE_LOG_RET(ERROR, OB_ERR_UNEXPECTED, "this ptr not in readable list",
KPC(removing_old_set_), KPC(readable_set_in_gc_));
readable_set_in_gc_ = nullptr;
} else if (readable_set_in_gc_->need_gc_scan(gc_start_time_)) {
// retry gc scan for OB_SIZE_OVERFLOW
} else {
READABLE_DLIST_FOREACH_X(readable_set_in_gc_, readable_tablets_list_,
!readable_set_in_gc_->need_gc_scan(gc_start_time_));
}
}
if (OB_ISNULL(readable_set_in_gc_)) {
if (!readable_tablets_list_.is_empty()) {
readable_set_in_gc_ = readable_tablets_list_.get_first();
new_round = true;
}
}
if (readable_set_in_gc_ == readable_tablets_list_.get_header()) {
if (readable_set_in_gc_ == readable_tablets_list_.get_header()
|| OB_ISNULL(readable_set_in_gc_)) {
readable_set_in_gc_ = nullptr;
DUP_TABLE_LOG(INFO, "readable_set_in_gc_ is null ptr, no need start gc", K(readable_tablets_list_.get_size()),
KP(removing_old_set_), KP(readable_set_in_gc_), KP(readable_tablets_list_.get_header()));
gc_start_time_ = ObTimeUtility::fast_current_time();
DUP_TABLE_LOG(INFO, "readable_set_in_gc_ is null ptr, no need start gc",
K(readable_tablets_list_.get_size()), KP(removing_old_set_),
KP(readable_set_in_gc_), KP(readable_tablets_list_.get_header()),
KP(readable_tablets_list_.get_first()));
}
return readable_set_in_gc_;
}
int ObLSDupTabletsMgr::scan_readable_set_for_gc()
int ObLSDupTabletsMgr::scan_readable_set_for_gc(const int64_t leader_takeover_ts)
{
int ret = OB_SUCCESS;
DupTabletChangeMap *tmp_readable_gc_set_ptr = nullptr;
bool new_round = false;
const int64_t last_scan_task_succ_ts =
MTL(ObTransService *)->get_dup_table_scan_task().get_last_scan_task_succ_ts();
SpinWLockGuard guard(dup_tablets_lock_);
if (OB_ISNULL(removing_old_set_)) {
@ -867,23 +880,27 @@ int ObLSDupTabletsMgr::scan_readable_set_for_gc()
} else if (removing_old_set_->get_related_set_op_type() <= DupTableRelatedSetOpType::INVALID) {
ret = OB_ERR_UNEXPECTED;
DUP_TABLE_LOG(WARN, "invalid related set type", K(ret), KPC(removing_old_set_));
} else if (leader_takeover_ts >= last_scan_task_succ_ts) {
ret = OB_EAGAIN;
DUP_TABLE_LOG(
WARN, "Since taking over, we have not yet scanned the schema of the dup table even once",
K(ret), K(leader_takeover_ts), K(last_scan_task_succ_ts),
K(MTL(ObTransService *)->get_dup_table_scan_task()));
} else if (removing_old_set_->get_related_set_op_type() == DupTableRelatedSetOpType::NEW_GC) {
if ((!removing_old_set_->empty())
|| removing_old_set_->get_related_common_header().is_valid()) {
if (is_busy_in_readable_change_()) {
// old set not confimed and clear, not need scan this time
} else if (OB_ISNULL(tmp_readable_gc_set_ptr = get_need_gc_set_(new_round))) {
} else if (OB_ISNULL(tmp_readable_gc_set_ptr = get_need_gc_set_())) {
// get null set
} else {
int64_t cur_time = ObTimeUtility::fast_current_time();
if (new_round || gc_start_time_ <= 0) {
gc_start_time_ = cur_time;
}
int64_t dup_schema_expired_time_interval =
3 * (MTL(ObTransService *)->get_dup_table_scan_task().get_max_exec_interval());
GcOneReadableSetHandler gc_handler(cur_time, tablet_gc_window_, *removing_old_set_,
MAX_GC_TABLET_COUNT);
GcOneReadableSetHandler gc_handler(gc_start_time_, dup_schema_expired_time_interval,
*removing_old_set_, MAX_GC_TABLET_COUNT);
if (OB_FAIL(hash_for_each_update(*tmp_readable_gc_set_ptr, gc_handler))) {
DUP_TABLE_LOG(WARN, "scan readable for gc failed", K(ret), KPC(removing_old_set_),
KPC(tmp_readable_gc_set_ptr), K(cur_time), K(gc_start_time_));
KPC(tmp_readable_gc_set_ptr), K(dup_schema_expired_time_interval),
K(gc_start_time_));
if (OB_SIZE_OVERFLOW == ret) {
ret = OB_SUCCESS;
}
@ -905,7 +922,7 @@ int ObLSDupTabletsMgr::scan_readable_set_for_gc()
DUP_TABLE_LOG(INFO, DUP_TABLET_LIFE_PREFIX "scan a readable set for GC", K(ret),
K(gc_handler.get_gc_tablet_cnt()), KPC(removing_old_set_),
KPC(tmp_readable_gc_set_ptr), K(new_round), K(gc_start_time_),
K(tablet_gc_window_));
K(dup_schema_expired_time_interval));
}
}
}
@ -913,6 +930,15 @@ int ObLSDupTabletsMgr::scan_readable_set_for_gc()
return ret;
}
bool ObLSDupTabletsMgr::is_busy_in_readable_change_()
{
bool is_busy =
// in gc
(!removing_old_set_->empty())
|| removing_old_set_->get_related_common_header().is_valid();
return is_busy;
}
// leader and follower remove tablet from readable set after first log synced
// lock outside this func
int ObLSDupTabletsMgr::remove_tablet_from_readable_set_()
@ -989,15 +1015,9 @@ int ObLSDupTabletsMgr::refresh_dup_tablet(const common::ObTabletID &tablet_id,
if (OB_FAIL(discover_dup_tablet_(tablet_id, refresh_time))) {
DUP_TABLE_LOG(WARN, "discover a dup tablet failed", K(tablet_id), K(refresh_time));
}
} else {
if (OB_NOT_NULL(removing_old_set_)
&& removing_old_set_->get_related_set_op_type() == DupTableRelatedSetOpType::OLD_GC) {
if (OB_FAIL(lose_dup_tablet_(tablet_id))) {
DUP_TABLE_LOG(WARN, "a dup tablet lose dup attr failed", K(tablet_id));
}
} else {
DUP_TABLE_LOG(WARN, "related set type not match", K(ret), KPC(removing_old_set_));
if (OB_FAIL(lose_dup_tablet_(tablet_id))) {
DUP_TABLE_LOG(WARN, "a dup tablet lose dup attr failed", K(tablet_id));
}
}
@ -1114,7 +1134,9 @@ OB_NOINLINE int ObLSDupTabletsMgr::process_prepare_ser_err_test_()
ret = EN_DUP_TABLE_LOG_PREPARE_SERIALIZE;
#endif
DUP_TABLE_LOG(INFO, "errsim prepare serialize err test", K(ret),K(ls_id_));
if (OB_FAIL(ret)) {
DUP_TABLE_LOG(INFO, "errsim for prepare serialize", K(ret), K(ls_id_));
}
return ret;
}
@ -1130,12 +1152,85 @@ int ObLSDupTabletsMgr::prepare_serialize(int64_t &max_ser_size,
unique_id_array.reuse();
if (OB_FAIL(process_prepare_ser_err_test_())) {
DUP_TABLE_LOG(WARN, "errsim for dup tablet log prepare serialize", K(ret));
}
if (OB_SUCC(ret)) {
if (OB_FAIL(process_prepare_ser_err_test_())) {
DUP_TABLE_LOG(WARN, "errsim for dup tablet log prepare serialize", K(ret));
if (OB_FAIL(prepare_serialize_confirming_tablet_set_(max_ser_size, unique_id_array,
max_log_buf_len))) {
DUP_TABLE_LOG(WARN, "prepare serialize confirming tablet set failed", K(ret), K(max_ser_size),
K(max_log_buf_len));
}
}
if (OB_SUCC(ret)) {
if (ObTimeUtility::fast_current_time() - ATOMIC_LOAD(&last_readable_sync_succ_time_)
<= MAX_READABLE_SET_SER_INTERVAL) {
DUP_TABLE_LOG(DEBUG,
"Too many readable tablets log entry. Stop serializing readable tablet log",
K(ret), K(unique_id_array), K(max_ser_size), K(last_readable_sync_succ_time_));
} else if (OB_FAIL(prepare_serialize_readable_tablet_set_(max_ser_size, unique_id_array,
max_log_buf_len))) {
DUP_TABLE_LOG(WARN, "prepare serialize readable tablet set failed", K(ret),
K(unique_id_array), K(max_ser_size), K(last_readable_sync_succ_time_));
}
}
if (OB_LOG_TOO_LARGE == ret) {
DUP_TABLE_LOG(DEBUG, "Too many dup tablets, we can not submit all", K(ret), K(max_ser_size),
K(max_log_buf_len), K(unique_id_array), K(unique_id_array.count()),
K(readable_tablets_list_.get_size()));
ret = OB_SUCCESS;
}
return ret;
}
int ObLSDupTabletsMgr::prepare_serialize_readable_tablet_set_(int64_t &max_ser_size,
DupTabletSetIDArray &unique_id_array,
const int64_t max_log_buf_len)
{
int ret = OB_SUCCESS;
// TODO serialize readable tablets
if (OB_SUCC(ret)) {
if (readable_tablets_list_.get_size() > 200) {
ret = OB_LOG_TOO_LARGE;
} else {
DLIST_FOREACH(readable_ptr, readable_tablets_list_)
{
if (readable_ptr->is_logging()) {
// set logging by old perpare serialize
DUP_TABLE_LOG(INFO, "this readable set is prepare serialized", K(ret),
K(readable_ptr->get_RO_common_header()));
} else if (OB_FAIL(cal_single_set_max_ser_size_(readable_ptr, max_ser_size, max_log_buf_len,
unique_id_array))) {
DUP_TABLE_LOG(WARN, "cal readable set max ser_size failed", K(ret));
if (OB_SIZE_OVERFLOW == ret) {
ret = OB_LOG_TOO_LARGE;
}
} else {
readable_ptr->set_logging();
}
}
DUP_TABLE_LOG(INFO, "serialize readable tablets in log for recovery", K(ret),
K(unique_id_array), K(max_ser_size), K(last_readable_sync_succ_time_),
K(last_readable_log_entry_scn_.atomic_load()));
}
}
return ret;
}
int ObLSDupTabletsMgr::prepare_serialize_confirming_tablet_set_(int64_t &max_ser_size,
DupTabletSetIDArray &unique_id_array,
const int64_t max_log_buf_len)
{
int ret = OB_SUCCESS;
int tmp_ret = OB_SUCCESS;
//need_confirm_queue
if (OB_SUCC(ret)) {
bool can_be_confirmed = true;
DLIST_FOREACH(cur_map, need_confirm_new_queue_)
@ -1164,6 +1259,7 @@ int ObLSDupTabletsMgr::prepare_serialize(int64_t &max_ser_size,
}
}
//changing new set
if (OB_SUCC(ret)) {
if (OB_ISNULL(changing_new_set_)) {
// do nothing
@ -1230,52 +1326,10 @@ int ObLSDupTabletsMgr::prepare_serialize(int64_t &max_ser_size,
}
}
}
if (OB_SUCC(ret)
&& ObTimeUtility::fast_current_time() - ATOMIC_LOAD(&last_readable_sync_succ_time_)
<= 30 * 1000 * 1000) {
ret = OB_LOG_TOO_LARGE;
DUP_TABLE_LOG(DEBUG,
"Too many readable tablets log entry. Stop serializing readable tablet log",
K(ret), K(unique_id_array), K(max_ser_size), K(last_readable_sync_succ_time_));
}
// TODO serialize readable tablets
if (OB_SUCC(ret)) {
if (readable_tablets_list_.get_size() > 200) {
ret = OB_LOG_TOO_LARGE;
} else {
DLIST_FOREACH(readable_ptr, readable_tablets_list_)
{
if (readable_ptr->is_logging()) {
// set logging by old perpare serialize
DUP_TABLE_LOG(INFO, "this readable set is prepare serialized", K(ret),
K(readable_ptr->get_RO_common_header()));
} else if (OB_FAIL(cal_single_set_max_ser_size_(readable_ptr, max_ser_size, max_log_buf_len,
unique_id_array))) {
DUP_TABLE_LOG(WARN, "cal readable set max ser_size failed", K(ret));
if (OB_SIZE_OVERFLOW == ret) {
ret = OB_LOG_TOO_LARGE;
}
} else {
readable_ptr->set_logging();
}
}
DUP_TABLE_LOG(INFO, "serialize readable tablets in log for recovery", K(ret),
K(unique_id_array), K(max_ser_size), K(last_readable_sync_succ_time_),
K(last_readable_log_entry_scn_.atomic_load()));
}
}
if (OB_LOG_TOO_LARGE == ret) {
DUP_TABLE_LOG(DEBUG, "Too many dup tablets, we can not submit all", K(ret), K(max_ser_size),
K(max_log_buf_len), K(unique_id_array), K(unique_id_array.count()),
K(readable_tablets_list_.get_size()));
ret = OB_SUCCESS;
}
return ret;
}
int ObLSDupTabletsMgr::serialize_tablet_log(const DupTabletSetIDArray &unique_id_array,
char *buf,
const int64_t buf_len,
@ -1589,7 +1643,7 @@ int ObLSDupTabletsMgr::tablet_log_synced(const bool sync_result,
}
if (unique_id_array.count() > 0) {
DUP_TABLE_LOG(DEBUG, "tablet log sync", K(ret), K(sync_result), K(for_replay), K(is_master()),
DUP_TABLE_LOG(INFO, "tablet log sync", K(ret), K(sync_result), K(for_replay), K(is_master()),
K(unique_id_array), K(scn), K(modify_readable_set));
}
@ -1771,6 +1825,19 @@ int64_t ObLSDupTabletsMgr::get_dup_tablet_count()
return total_size;
}
int64_t ObLSDupTabletsMgr::get_readable_tablet_count()
{
SpinRLockGuard guard(dup_tablets_lock_);
int64_t total_size = 0;
DLIST_FOREACH_X(readable_set_ptr, readable_tablets_list_, true)
{
total_size += readable_set_ptr->size();
}
return total_size;
}
bool ObLSDupTabletsMgr::has_dup_tablet() { return 0 < get_dup_tablet_count(); }
int64_t ObLSDupTabletsMgr::get_readable_tablet_set_count()
@ -2139,6 +2206,7 @@ int ObLSDupTabletsMgr::lose_dup_tablet_(const common::ObTabletID &tablet_id)
* In changing_new: remove from new set
* */
DupTabletInfo tmp_info;
bool remove_from_temporary = false;
DupTabletSetCommonHeader changing_new_header;
changing_new_header.set_invalid_unique_id();
@ -2155,6 +2223,8 @@ int ObLSDupTabletsMgr::lose_dup_tablet_(const common::ObTabletID &tablet_id)
} else if (OB_SUCC(changing_new_map->get_refactored(tablet_id, tmp_info))) {
if (OB_FAIL(changing_new_map->erase_refactored(tablet_id))) {
DUP_TABLE_LOG(WARN, "remove from changing_new_set_ failed", K(ret), K(tablet_id));
} else {
remove_from_temporary = true;
}
} else if (ret != OB_HASH_NOT_EXIST) {
DUP_TABLE_LOG(WARN, "get dup table status from new_tablets_ failed", K(ret));
@ -2162,40 +2232,30 @@ int ObLSDupTabletsMgr::lose_dup_tablet_(const common::ObTabletID &tablet_id)
ret = OB_SUCCESS;
}
DupTabletSetCommonHeader old_set_header;
old_set_header.set_invalid_unique_id();
old_set_header.set_old();
DupTabletChangeMap *old_tablet_set = nullptr;
if (OB_FAIL(ret)) {
// do nothing
} else if (OB_FAIL(get_target_tablet_set_(old_set_header, old_tablet_set))) {
DUP_TABLE_LOG(WARN, "get old tablets set failed", K(ret));
} else if (!old_tablet_set->get_change_status()->is_modifiable()) {
ret = OB_SUCCESS;
DUP_TABLE_LOG(DEBUG, "old tablet set can not be modified", K(ret), K(tablet_id),
KPC(old_tablet_set));
} else {
DLIST_FOREACH(readable_set_ptr, readable_tablets_list_)
if (OB_SUCC(ret) && remove_from_temporary) {
DUP_TABLE_LOG(INFO, DUP_TABLET_LIFE_PREFIX "lose a temporary dup tablet", K(ret), K(tablet_id),
KPC(changing_new_set_));
}
#ifdef ENABLE_DEBUG_LOG
if (OB_SUCC(ret) && !remove_from_temporary) {
ret = OB_HASH_NOT_EXIST;
DLIST_FOREACH_X(readable_set_ptr, readable_tablets_list_, OB_HASH_NOT_EXIST == ret)
{
if (readable_set_ptr->is_logging()) {
// do nothing
} else if (OB_SUCC(readable_set_ptr->get_refactored(tablet_id, tmp_info))) {
if (OB_FAIL(old_tablet_set->set_refactored(tablet_id, tmp_info))) {
DUP_TABLE_LOG(WARN, "insert into old tablet set failed", K(ret));
} else if (OB_FAIL(readable_set_ptr->erase_refactored(tablet_id))) {
DUP_TABLE_LOG(WARN, "remove from readable tablet set failed", K(ret));
}
if (OB_SUCC(readable_set_ptr->get_refactored(tablet_id, tmp_info))) {
DUP_TABLE_LOG(INFO, "lose a dup tablet in readable tabelt set which need be gc", K(ret),
K(tablet_id), K(tmp_info), KPC(readable_set_ptr));
} else if (ret != OB_HASH_NOT_EXIST) {
DUP_TABLE_LOG(WARN, "get dup table status from readable_tablets_ failed", K(ret));
} else if (ret == OB_HASH_NOT_EXIST) {
ret = OB_SUCCESS;
DUP_TABLE_LOG(WARN, "get dup table status from readable_tablets_ failed", K(ret),
KPC(readable_set_ptr));
}
}
if (ret == OB_HASH_NOT_EXIST) {
ret = OB_SUCCESS;
}
}
if (OB_FAIL(ret)) {
DUP_TABLE_LOG(WARN, "lose dup tablet failed", K(ret), K(tablet_id), KPC(changing_new_set_),
KPC(removing_old_set_), K(readable_tablets_list_.get_size()));
}
#endif
return ret;
}
@ -3185,6 +3245,8 @@ int ObTenantDupTabletSchemaHelper::refresh_and_get_tablet_set(TabletIDSet &tenan
{
int ret = OB_SUCCESS;
tenant_dup_tablet_set.reuse();
if (OB_FAIL(get_all_dup_tablet_set_(tenant_dup_tablet_set))) {
DUP_TABLE_LOG(WARN, "get tenant dup tablet set faild", K(ret));
}

29
src/storage/tx/ob_dup_table_tablets.h
View File

@ -419,10 +419,6 @@ public:
DUP_TABLE_LOG(INFO, "this readable set used for other operation, should skip gc", KPC(this));
bool_ret = false;
} else {
bool_ret = true;
}
if (bool_ret) {
if (last_gc_scan_ts_ <= 0
|| gc_start_time > last_gc_scan_ts_) {
bool_ret = true;
@ -431,6 +427,8 @@ public:
}
}
DUP_TABLE_LOG(INFO, "check need gc scan", K(bool_ret),K(last_gc_scan_ts_),K(gc_start_time),KPC(this));
return bool_ret;
}
void set_last_gc_scan_ts(const int64_t gc_start_time) {
@ -440,6 +438,8 @@ public:
} else {
last_gc_scan_ts_ = gc_start_time;
}
DUP_TABLE_LOG(INFO, "set last gc scn ts", K(last_gc_scan_ts_),K(gc_start_time),KPC(this));
}
int64_t get_last_gc_scan_ts() { return last_gc_scan_ts_; }
@ -629,7 +629,7 @@ public:
const share::SCN &to_scn);
int gc_tmporary_dup_tablets(const int64_t gc_ts, const int64_t max_task_interval);
// new gc methods
int scan_readable_set_for_gc();
int scan_readable_set_for_gc(const int64_t leader_takeover_ts);
int refresh_dup_tablet(const common::ObTabletID &tablet_id,
bool is_dup_table,
@ -662,6 +662,7 @@ public:
int try_to_confirm_tablets(const share::SCN &confirm_scn);
// bool need_log_tablets();
int64_t get_dup_tablet_count();
int64_t get_readable_tablet_count();
bool has_dup_tablet();
int64_t get_readable_tablet_set_count();
int64_t get_need_confirm_tablet_set_count();
@ -835,6 +836,13 @@ private:
const bool for_replay);
bool need_seralize_readable_set() { return true; }
int prepare_serialize_readable_tablet_set_(int64_t &max_ser_size,
DupTabletSetIDArray &unique_id_array,
const int64_t max_log_buf_len);
int prepare_serialize_confirming_tablet_set_(int64_t &max_ser_size,
DupTabletSetIDArray &unique_id_array,
const int64_t max_log_buf_len);
int cal_single_set_max_ser_size_(DupTabletChangeMap *hash_map,
int64_t &max_ser_size,
const int64_t ser_size_limit,
@ -854,8 +862,9 @@ private:
int remove_src_and_related_set_header_from_array_(DupTabletChangeMap *src_set,
DupTabletChangeMap *related_set,
DupTabletSetIDArray &unique_id_array);
DupTabletChangeMap *get_need_gc_set_(bool &new_round);
DupTabletChangeMap *get_need_gc_set_();
int remove_tablet_from_readable_set_();
bool is_busy_in_readable_change_();
private:
//
static int64_t GC_DUP_TABLETS_TIME_INTERVAL; // 5 min
@ -866,6 +875,7 @@ private:
const static int64_t MAX_FREE_SET_COUNT;
const static int64_t MAX_GC_TABLET_COUNT;
static int64_t MAX_READABLE_SET_SER_INTERVAL;
public:
TO_STRING_KV(K(free_set_pool_.get_size()),
@ -873,7 +883,6 @@ public:
K(need_confirm_new_queue_.get_size()),
K(readable_tablets_list_.get_size()),
KPC(removing_old_set_),
K(last_gc_succ_time_),
K(last_no_free_set_time_),
K(extra_free_set_alloc_count_));
@ -885,16 +894,12 @@ private:
bool is_master_;
bool is_stopped_;
// used for gc_handler
int64_t tablet_gc_window_; // default is 2 * ObDupTabletScanTask::DUP_TABLET_SCAN_INTERVAL;
common::ObDList<DupTabletChangeMap> free_set_pool_;
DupTabletChangeMap *changing_new_set_;
common::ObDList<DupTabletChangeMap> need_confirm_new_queue_;
common::ObDList<DupTabletChangeMap> readable_tablets_list_;
DupTabletChangeMap *removing_old_set_;
// gc_dup_table
int64_t last_gc_succ_time_;
/* 1. gc one round means iter all readable set
* 2. use readable_set_in_gc_ point to readable set not finish gc in one round
* 3. use gc_start_time_ mark gc one round start time one round

18
src/storage/tx/ob_dup_table_ts_sync.cpp
View File

@ -355,13 +355,23 @@ int ObDupTableLSTsSyncMgr::get_local_ts_info(DupTableTsInfo &ts_info)
int ObDupTableLSTsSyncMgr::get_cache_ts_info(const common::ObAddr &addr, DupTableTsInfo &ts_info)
{
int ret = OB_SUCCESS;
int tmp_ret = OB_SUCCESS;
SpinRLockGuard guard(ts_sync_lock_);
{
SpinRLockGuard guard(ts_sync_lock_);
if (OB_FAIL(get_ts_info_cache_(addr, ts_info))) {
DUP_TABLE_LOG(WARN, "get ts info cache failed", K(ret));
if (OB_FAIL(get_ts_info_cache_(addr, ts_info))) {
DUP_TABLE_LOG(WARN, "get ts info cache failed", K(ret));
}
DUP_TABLE_LOG(DEBUG, "get ts info cache", K(ret), K(ret), K(ts_info));
}
if (OB_HASH_NOT_EXIST == ret) {
if (OB_TMP_FAIL(request_ts_info(addr))) {
DUP_TABLE_LOG(WARN, "request ts info failed", K(tmp_ret), K(ts_info));
}
}
DUP_TABLE_LOG(DEBUG, "get ts info cache", K(ret), K(ret), K(ts_info));
return ret;
}

290
src/storage/tx/ob_dup_table_util.cpp
View File

@ -38,8 +38,13 @@ void ObDupTabletScanTask::reset()
tenant_id_ = 0;
dup_table_scan_timer_ = nullptr;
dup_loop_worker_ = nullptr;
last_execute_time_ = 0;
max_execute_interval_ = 0;
min_dup_ls_status_info_.reset();
tenant_schema_dup_tablet_set_.destroy();
scan_task_execute_interval_ = ObDupTabletScanTask::DUP_TABLET_SCAN_INTERVAL;
last_dup_ls_refresh_time_ = 0;
last_dup_schema_refresh_time_ = 0;
last_scan_task_succ_time_ = 0;
max_execute_interval_ = ObDupTabletScanTask::DUP_TABLET_SCAN_INTERVAL;
}
int ObDupTabletScanTask::make(const int64_t tenant_id,
@ -54,6 +59,9 @@ int ObDupTabletScanTask::make(const int64_t tenant_id,
tenant_id_ = tenant_id;
dup_table_scan_timer_ = scan_timer;
dup_loop_worker_ = loop_worker;
min_dup_ls_status_info_.reset();
tenant_schema_dup_tablet_set_.reuse();
scan_task_execute_interval_ = ObDupTabletScanTask::DUP_TABLET_SCAN_INTERVAL;
// ObTransTask::make(ObTransRetryTaskType::DUP_TABLET_SCAN_TASK);
// set_retry_interval_us(DUP_TABLET_SCAN_INTERVAL, DUP_TABLET_SCAN_INTERVAL);
}
@ -74,83 +82,143 @@ void ObDupTabletScanTask::runTimerTask()
}
dup_table_scan_timer_->unregister_timeout_task(*this);
dup_table_scan_timer_->register_timeout_task(*this, DUP_TABLET_SCAN_INTERVAL);
dup_table_scan_timer_->register_timeout_task(*this, scan_task_execute_interval_);
}
}
int ObDupTabletScanTask::refresh_dup_tablet_schema_(
bool need_refresh,
ObTenantDupTabletSchemaHelper::TabletIDSet &tenant_dup_tablet_set,
share::ObLSStatusInfo &dup_ls_status_info)
int ObDupTabletScanTask::refresh_dup_ls_(const int64_t cur_time)
{
int ret = OB_SUCCESS;
bool has_dup_ls = false;
if (need_refresh) {
bool need_refresh = true;
if (min_dup_ls_status_info_.is_valid() && min_dup_ls_status_info_.is_duplicate_ls()) {
if (cur_time - last_dup_ls_refresh_time_ < MAX_DUP_LS_REFRESH_INTERVAL) {
need_refresh = false;
} else {
need_refresh = true;
}
} else {
need_refresh = true;
}
if (need_refresh && OB_SUCC(ret)) {
share::ObLSStatusInfo tmp_dup_ls_status_info;
share::ObLSStatusOperator ls_status_op;
if (OB_FAIL(ls_status_op.get_duplicate_ls_status_info(MTL_ID(), *GCTX.sql_proxy_,
dup_ls_status_info, share::OBCG_STORAGE))) {
tmp_dup_ls_status_info, share::OBCG_STORAGE))) {
if (OB_ENTRY_NOT_EXIST == ret) {
DUP_TABLE_LOG(DEBUG, "no duplicate ls", K(dup_ls_status_info));
DUP_TABLE_LOG(DEBUG, "no duplicate ls", K(tmp_dup_ls_status_info));
ret = OB_SUCCESS;
} else {
DUP_TABLE_LOG(WARN, "get duplicate ls status info failed", K(ret), K(dup_ls_status_info));
DUP_TABLE_LOG(WARN, "get duplicate ls status info failed", K(ret),
K(tmp_dup_ls_status_info));
}
} else {
DUP_TABLE_LOG(INFO, "find a duplicate ls", K(ret), K(dup_ls_status_info));
}
DUP_TABLE_LOG(DEBUG, "find a duplicate ls", K(ret), K(tmp_dup_ls_status_info));
if (OB_SUCC(ret) && dup_ls_status_info.is_duplicate_ls()) {
if (OB_FAIL(ret)) {
// do nothing
} else if (!tenant_dup_tablet_set.created()) {
if (OB_FAIL(tenant_dup_tablet_set.create(512))) {
DUP_TABLE_LOG(WARN, "init dup tablet cache failed", K(ret));
if (!tmp_dup_ls_status_info.is_valid() || !tmp_dup_ls_status_info.is_duplicate_ls()) {
ret = OB_ERR_UNEXPECTED;
DUP_TABLE_LOG(ERROR, "invalid tmp_dup_ls_status_info", K(ret), K(tmp_dup_ls_status_info));
} else {
if (min_dup_ls_status_info_.is_valid() && min_dup_ls_status_info_.is_duplicate_ls()
&& tmp_dup_ls_status_info.get_ls_id() != min_dup_ls_status_info_.get_ls_id()) {
DUP_TABLE_LOG(ERROR, "The min_dup_ls has already changed", K(ret),
K(tmp_dup_ls_status_info), K(min_dup_ls_status_info_));
}
if (OB_FAIL(min_dup_ls_status_info_.assign(tmp_dup_ls_status_info))) {
DUP_TABLE_LOG(WARN, "rewrite min_dup_ls_status_info_ failed", K(ret),
K(tmp_dup_ls_status_info), K(min_dup_ls_status_info_));
} else {
last_dup_ls_refresh_time_ = cur_time;
}
}
if (OB_FAIL(ret)) {
// do nothing
} else if (OB_FAIL(dup_schema_helper_.refresh_and_get_tablet_set(tenant_dup_tablet_set))) {
DUP_TABLE_LOG(WARN, "refresh dup tablet set failed", K(ret));
}
}
}
return ret;
}
int ObDupTabletScanTask::refresh_dup_tablet_schema_(const int64_t cur_time)
{
int ret = OB_SUCCESS;
if (OB_SUCC(ret) && min_dup_ls_status_info_.is_valid()
&& min_dup_ls_status_info_.is_duplicate_ls()) {
if (OB_FAIL(ret)) {
// do nothing
} else if (!tenant_schema_dup_tablet_set_.created()) {
if (OB_FAIL(tenant_schema_dup_tablet_set_.create(512))) {
DUP_TABLE_LOG(WARN, "init dup tablet cache failed", K(ret));
}
}
if (OB_FAIL(ret)) {
// do nothing
} else if (cur_time <= last_dup_schema_refresh_time_) {
// do nothing
} else if (OB_FAIL(
dup_schema_helper_.refresh_and_get_tablet_set(tenant_schema_dup_tablet_set_))) {
DUP_TABLE_LOG(WARN, "refresh dup tablet set failed", K(ret));
tenant_schema_dup_tablet_set_.clear();
} else {
last_dup_schema_refresh_time_ = cur_time;
}
}
return ret;
}
bool ObDupTabletScanTask::has_valid_dup_schema_() const
{
bool dup_schema_is_valid = false;
dup_schema_is_valid = min_dup_ls_status_info_.is_valid()
&& min_dup_ls_status_info_.is_duplicate_ls()
&& !tenant_schema_dup_tablet_set_.empty();
return dup_schema_is_valid;
}
int ObDupTabletScanTask::execute_for_dup_ls_()
{
int ret = OB_SUCCESS;
int tmp_ret = OB_SUCCESS;
TabletIDArray tablet_id_array;
ObTenantDupTabletSchemaHelper::TabletIDSet tenant_dup_tablet_set;
bool need_refreh_dup_schema = true;
ObLSHandle ls_handle;
share::ObLSStatusInfo dup_ls_status_info;
// compute scan task max execute interval
const int64_t cur_time = ObTimeUtility::fast_current_time();
if (cur_time - last_execute_time_ > 0) {
if (0 != last_execute_time_) {
max_execute_interval_ = max(max_execute_interval_, cur_time - last_execute_time_);
last_execute_time_ = cur_time;
if (cur_time - last_scan_task_succ_time_ > 0) {
if (0 != last_scan_task_succ_time_) {
if (max_execute_interval_ / 2 >= (cur_time - last_scan_task_succ_time_)) {
// Avoid residual excessive execution intervals in exceptional circumstances
ATOMIC_STORE(&max_execute_interval_, cur_time - last_scan_task_succ_time_);
} else {
ATOMIC_STORE(&max_execute_interval_,
max(max_execute_interval_, cur_time - last_scan_task_succ_time_));
}
} else {
last_execute_time_ = ObTimeUtility::fast_current_time();
}
}
if (OB_TMP_FAIL(refresh_dup_ls_(cur_time))) {
DUP_TABLE_LOG(WARN, "refresh dup ls failed", K(tmp_ret), KPC(this));
} else if (OB_TMP_FAIL(refresh_dup_tablet_schema_(cur_time))) {
DUP_TABLE_LOG(WARN, "refresh dup schema failed", K(tmp_ret), KPC(this));
}
if (OB_ISNULL(MTL(ObLSService *)) || OB_ISNULL(dup_loop_worker_)) {
ret = OB_INVALID_ARGUMENT;
DUP_TABLE_LOG(WARN, "invalid arguments", K(ret));
} else if (OB_FAIL(refresh_dup_tablet_schema_(need_refreh_dup_schema, tenant_dup_tablet_set,
dup_ls_status_info))) {
DUP_TABLE_LOG(WARN, "refresh dup table schema failed", K(ret));
} else if (!dup_ls_status_info.is_duplicate_ls()) {
} else if (!has_valid_dup_schema_()) {
DUP_TABLE_LOG(DEBUG, "refresh dup table schema failed", K(ret), KPC(this));
// do nothing
} else if (OB_FAIL(MTL(ObLSService *)
->get_ls(dup_ls_status_info.ls_id_, ls_handle, ObLSGetMod::TRANS_MOD))) {
DUP_TABLE_LOG(WARN, "get dup ls failed", K(ret), K(dup_ls_status_info));
} else if (OB_FAIL(
MTL(ObLSService *)
->get_ls(min_dup_ls_status_info_.ls_id_, ls_handle, ObLSGetMod::TRANS_MOD))) {
DUP_TABLE_LOG(WARN, "get dup ls failed", K(ret), KPC(this));
} else {
ObLS *cur_ls_ptr = ls_handle.get_ls();
@ -158,16 +226,9 @@ int ObDupTabletScanTask::execute_for_dup_ls_()
ret = OB_INVALID_ARGUMENT;
DUP_TABLE_LOG(WARN, "invalid ls ptr", K(ret), KP(cur_ls_ptr));
} else if (!cur_ls_ptr->get_dup_table_ls_handler()->is_master()) {
// #ifndef NDEBUG
DUP_TABLE_LOG(INFO,
"ls not leader",
K(cur_ls_ptr->get_ls_id()));
// #endif
} else if (OB_FAIL(refresh_dup_tablet_schema_(need_refreh_dup_schema, tenant_dup_tablet_set,
dup_ls_status_info))) {
DUP_TABLE_LOG(INFO, "refresh dup table schema failed", K(ret));
} else if (OB_FALSE_IT(need_refreh_dup_schema = false)) {
// do nothing
#ifndef NDEBUG
DUP_TABLE_LOG(INFO, "ls not leader", K(cur_ls_ptr->get_ls_id()));
#endif
} else {
storage::ObHALSTabletIDIterator ls_tablet_id_iter(cur_ls_ptr->get_ls_id(), true);
if (OB_FAIL(cur_ls_ptr->build_tablet_iter(ls_tablet_id_iter))) {
@ -180,7 +241,7 @@ int ObDupTabletScanTask::execute_for_dup_ls_()
int64_t refresh_time = ObTimeUtility::fast_current_time();
while (OB_SUCC(ls_tablet_id_iter.get_next_tablet_id(tmp_tablet_id))) {
is_dup_tablet = false;
ret = tenant_dup_tablet_set.exist_refactored(tmp_tablet_id);
ret = tenant_schema_dup_tablet_set_.exist_refactored(tmp_tablet_id);
if (OB_HASH_EXIST == ret) {
is_dup_tablet = true;
ret = OB_SUCCESS;
@ -193,7 +254,9 @@ int ObDupTabletScanTask::execute_for_dup_ls_()
K(ret), K(cur_ls_ptr->get_ls_id()), K(tmp_tablet_id));
}
if (!cur_ls_ptr->get_dup_table_ls_handler()->is_inited() && !is_dup_tablet) {
if (OB_FAIL(ret)) {
// do nothing
} else if (!cur_ls_ptr->get_dup_table_ls_handler()->is_inited() && !is_dup_tablet) {
// do nothing
} else if (OB_FAIL(cur_ls_ptr->get_dup_table_ls_handler()->init(is_dup_tablet))
&& OB_INIT_TWICE != ret) {
@ -210,31 +273,31 @@ int ObDupTabletScanTask::execute_for_dup_ls_()
}
if (OB_ITER_END == ret) {
// ret = OB_SUCCESS;
if (OB_FAIL(cur_ls_ptr->get_dup_table_ls_handler()->gc_temporary_dup_tablets(
refresh_time, max_execute_interval_))) {
DUP_TABLE_LOG(WARN, "ls gc dup_tablet failed", KR(ret), K(refresh_time),
K(max_execute_interval_));
}
ret = OB_SUCCESS;
}
}
}
// refresh dup_table_ls on leader and follower
if (!cur_ls_ptr->get_dup_table_ls_handler()->check_tablet_set_exist()) {
// refresh dup_table_ls on leader and follower
if (!cur_ls_ptr->get_dup_table_ls_handler()->is_inited()
|| !cur_ls_ptr->get_dup_table_ls_handler()->check_tablet_set_exist()) {
// do nothing
} else if (OB_FAIL(dup_loop_worker_->append_dup_table_ls(cur_ls_ptr->get_ls_id()))) {
} else if (OB_TMP_FAIL(dup_loop_worker_->append_dup_table_ls(cur_ls_ptr->get_ls_id()))) {
DUP_TABLE_LOG(WARN, "refresh dup_table ls failed", K(ret));
}
}
if (tenant_dup_tablet_set.created()) {
tenant_dup_tablet_set.destroy();
}
if (OB_FAIL(ret)) {
DUP_TABLE_LOG(WARN, "scan dup ls to find dup_tablet failed", KR(ret));
} else {
ATOMIC_STORE(&last_scan_task_succ_time_, cur_time);
}
#ifndef NDEBUG
DUP_TABLE_LOG(INFO, "execute dup table scan task", KPC(this));
#else
#endif
return ret;
}
@ -453,15 +516,6 @@ int ObDupTableLSHandler::safe_to_destroy(bool &is_dup_table_handler_safe)
void ObDupTableLSHandler::destroy() { reset(); }
// int ObDupTableLSHandler::offline()
// {
// int ret = OB_SUCCESS;
// }
// return ret;
// }
//
// int ObDupTableLSHandler::online() {}
void ObDupTableLSHandler::reset()
{
// ATOMIC_STORE(&is_inited_, false);
@ -502,38 +556,6 @@ void ObDupTableLSHandler::reset()
}
}
// bool ObDupTableLSHandler::is_master()
// {
// bool sub_master = true;
// if (OB_NOT_NULL(ts_sync_mgr_ptr_)) {
// sub_master = sub_master && ts_sync_mgr_ptr_->is_master();
// }
// if (OB_NOT_NULL(lease_mgr_ptr_)) {
// sub_master = sub_master && lease_mgr_ptr_->is_master();
// }
// if (OB_NOT_NULL(tablets_mgr_ptr_)) {
// sub_master = sub_master && tablets_mgr_ptr_->is_master();
// }
//
// return (ATOMIC_LOAD(&is_master_)) && sub_master;
// }
// bool ObDupTableLSHandler::is_follower()
// {
// bool sub_not_master = true;
// if (OB_NOT_NULL(ts_sync_mgr_ptr_)) {
// sub_not_master = sub_not_master && !ts_sync_mgr_ptr_->is_master();
// }
// if (OB_NOT_NULL(lease_mgr_ptr_)) {
// sub_not_master = sub_not_master && !lease_mgr_ptr_->is_master();
// }
// if (OB_NOT_NULL(tablets_mgr_ptr_)) {
// sub_not_master = sub_not_master && !tablets_mgr_ptr_->is_master();
// }
//
// return (!ATOMIC_LOAD(&is_master_)) && sub_not_master;
// }
bool ObDupTableLSHandler::is_inited()
{
SpinRLockGuard r_init_guard(init_rw_lock_);
@ -542,21 +564,6 @@ bool ObDupTableLSHandler::is_inited()
bool ObDupTableLSHandler::is_master() { return ls_state_helper_.is_leader_serving(); }
// bool ObDupTableLSHandler::is_online()
// {
// bool sub_online = true;
// if (OB_NOT_NULL(ts_sync_mgr_ptr_)) {
// sub_online = sub_online && !ts_sync_mgr_ptr_->is_master();
// }
// if (OB_NOT_NULL(lease_mgr_ptr_)) {
// sub_online = sub_online && !lease_mgr_ptr_->is_master();
// }
// if (OB_NOT_NULL(tablets_mgr_ptr_)) {
// sub_online = sub_online && !tablets_mgr_ptr_->is_master();
// }
//
// }
int ObDupTableLSHandler::ls_loop_handle()
{
int ret = OB_SUCCESS;
@ -581,19 +588,27 @@ int ObDupTableLSHandler::ls_loop_handle()
if (OB_ISNULL(log_operator_) || !log_operator_->is_busy()) {
// handle lease request and collect follower info
DupTableTsInfo min_lease_ts_info;
if (OB_FAIL(get_min_lease_ts_info_(min_lease_ts_info))) {
DUP_TABLE_LOG(WARN, "get min lease ts info failed", K(ret), K(min_lease_ts_info));
if (OB_TMP_FAIL(get_min_lease_ts_info_(min_lease_ts_info))) {
DUP_TABLE_LOG(WARN, "get min lease ts info failed", K(tmp_ret), K(min_lease_ts_info));
// try confirm tablets and check tablet need log
} else if (OB_FAIL(try_to_confirm_tablets_(min_lease_ts_info.max_replayed_scn_))) {
DUP_TABLE_LOG(WARN, "try confirm tablets failed", K(ret), K(min_lease_ts_info));
} else {
// submit lease log
} else if (OB_TMP_FAIL(try_to_confirm_tablets_(min_lease_ts_info.max_replayed_scn_))) {
DUP_TABLE_LOG(WARN, "try confirm tablets failed", K(tmp_ret), K(min_lease_ts_info));
}
if (OB_TMP_FAIL(tablets_mgr_ptr_->scan_readable_set_for_gc(
ATOMIC_LOAD(&interface_stat_.dup_table_ls_leader_takeover_ts_)))) {
DUP_TABLE_LOG(WARN, "scan readable set failed", K(tmp_ret));
}
// submit lease log
if(OB_FAIL(ret))
{
} else
if (OB_ISNULL(log_operator_)) {
ret = OB_INVALID_ARGUMENT;
DUP_TABLE_LOG(WARN, "invalid log operator ptr", K(ret), KP(log_operator_));
} else if (OB_FAIL(log_operator_->submit_log_entry())) {
DUP_TABLE_LOG(WARN, "submit dup table log entry failed", K(ret));
}
} else if (OB_FAIL(log_operator_->submit_log_entry())) {
DUP_TABLE_LOG(WARN, "submit dup table log entry failed", K(ret));
}
}
@ -1359,6 +1374,9 @@ int ObDupTableLSHandler::switch_to_leader()
if (OB_FAIL(ls_state_helper_.state_change_succ(ObDupTableLSRoleState::LS_TAKEOVER_SUCC,
restore_state_container))) {
DUP_TABLE_LOG(ERROR, "change ls role state error", K(ret), KPC(this));
} else {
ATOMIC_STORE(&interface_stat_.dup_table_ls_leader_takeover_ts_,
ObTimeUtility::fast_current_time());
}
} else {
tmp_ret = OB_SUCCESS;
@ -1580,6 +1598,8 @@ int ObDupTableLSHandler::get_min_lease_ts_info_(DupTableTsInfo &min_ts_info)
{
int ret = OB_SUCCESS;
int ts_info_err_ret = OB_SUCCESS;
LeaseAddrArray lease_valid_array;
min_ts_info.reset();
@ -1599,6 +1619,10 @@ int ObDupTableLSHandler::get_min_lease_ts_info_(DupTableTsInfo &min_ts_info)
for (int64_t i = 0; OB_SUCC(ret) && i < lease_valid_array.count(); i++) {
if (OB_FAIL(ts_sync_mgr_ptr_->get_cache_ts_info(lease_valid_array[i], tmp_ts_info))) {
DUP_TABLE_LOG(WARN, "get cache ts info failed", K(ret), K(lease_valid_array[i]));
if (OB_HASH_NOT_EXIST == ret) {
ts_info_err_ret = ret;
ret = OB_SUCCESS;
}
} else {
min_ts_info.max_replayed_scn_ =
share::SCN::min(min_ts_info.max_replayed_scn_, tmp_ts_info.max_replayed_scn_);
@ -1610,8 +1634,12 @@ int ObDupTableLSHandler::get_min_lease_ts_info_(DupTableTsInfo &min_ts_info)
}
}
if (OB_SUCC(ret) && ts_info_err_ret != OB_SUCCESS) {
ret = ts_info_err_ret;
}
if (OB_FAIL(ret)) {
DUP_TABLE_LOG(INFO, "get min lease ts info failed", K(ret), K(min_ts_info),
DUP_TABLE_LOG(WARN, "get min lease ts info failed", K(ret), K(ts_info_err_ret), K(min_ts_info),
K(lease_valid_array));
}
return ret;

36
src/storage/tx/ob_dup_table_util.h
View File

@ -17,6 +17,7 @@
#include "storage/tx/ob_dup_table_lease.h"
#include "storage/tx/ob_trans_define.h"
#include "storage/tx/ob_dup_table_stat.h"
#include "share/ls/ob_ls_status_operator.h"
namespace oceanbase
{
@ -57,7 +58,8 @@ public:
class ObDupTabletScanTask : public ObITimeoutTask
{
public:
static const int64_t DUP_TABLET_SCAN_INTERVAL = 10 * 1000 * 1000; // 10s
static const int64_t DUP_TABLET_SCAN_INTERVAL = 10 * 1000 * 1000; // 10s
static const int64_t MAX_DUP_LS_REFRESH_INTERVAL = 60 * 60 * 1000 * 1000L; // 60min
public:
ObDupTabletScanTask() { reset(); }
~ObDupTabletScanTask() { destroy(); }
@ -70,11 +72,26 @@ public:
void runTimerTask();
uint64_t hash() const { return tenant_id_; }
int64_t get_max_exec_interval() const { return ATOMIC_LOAD(&max_execute_interval_); }
int64_t get_last_scan_task_succ_ts() const { return ATOMIC_LOAD(&last_scan_task_succ_time_); }
TO_STRING_KV(K(tenant_id_),
KP(dup_table_scan_timer_),
KP(dup_loop_worker_),
K(min_dup_ls_status_info_),
K(tenant_schema_dup_tablet_set_.size()),
K(scan_task_execute_interval_),
K(last_dup_ls_refresh_time_),
K(last_dup_schema_refresh_time_),
K(last_scan_task_succ_time_),
K(max_execute_interval_));
private:
int execute_for_dup_ls_();
int refresh_dup_tablet_schema_(bool need_refresh,
ObTenantDupTabletSchemaHelper::TabletIDSet &tenant_dup_tablet_set,
share::ObLSStatusInfo &dup_ls_status_info);
int refresh_dup_ls_(const int64_t cur_time);
int refresh_dup_tablet_schema_(const int64_t cur_time);
bool has_valid_dup_schema_() const;
private:
ObTenantDupTabletSchemaHelper dup_schema_helper_;
@ -82,7 +99,16 @@ private:
int64_t tenant_id_;
ObDupTableLeaseTimer *dup_table_scan_timer_;
ObDupTableLoopWorker *dup_loop_worker_;
int64_t last_execute_time_;
share::ObLSStatusInfo min_dup_ls_status_info_; // min_ls_id
ObTenantDupTabletSchemaHelper::TabletIDSet tenant_schema_dup_tablet_set_;
int64_t scan_task_execute_interval_;
int64_t last_dup_ls_refresh_time_;
int64_t last_dup_schema_refresh_time_;
int64_t last_scan_task_succ_time_;
int64_t max_execute_interval_;
};

1
src/storage/tx/ob_trans_service.h
View File

@ -230,6 +230,7 @@ public:
ObIDupTableRpc *get_dup_table_rpc() { return dup_table_rpc_; }
ObDupTableRpc &get_dup_table_rpc_impl() { return dup_table_rpc_impl_; }
ObDupTableLoopWorker &get_dup_table_loop_worker() { return dup_table_loop_worker_; }
const ObDupTabletScanTask &get_dup_table_scan_task() { return dup_tablet_scan_task_; }
ObILocationAdapter *get_location_adapter() { return location_adapter_; }
common::ObMySQLProxy *get_mysql_proxy() { return GCTX.sql_proxy_; }
bool is_running() const { return is_running_; }