471 lines
14 KiB
C++
471 lines
14 KiB
C++
// owner: zjf225077
|
|
// owner group: log
|
|
|
|
/**
|
|
* 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 private public
|
|
#include "env/ob_simple_log_cluster_env.h"
|
|
#undef private
|
|
|
|
const std::string TEST_NAME = "apply_func";
|
|
using namespace oceanbase::common;
|
|
using namespace oceanbase;
|
|
|
|
namespace oceanbase
|
|
{
|
|
using namespace logservice;
|
|
namespace unittest
|
|
{
|
|
class MockLSAdapter : public ObLSAdapter
|
|
{
|
|
public:
|
|
MockLSAdapter() {
|
|
ObLSAdapter();
|
|
success_count_ = 0;
|
|
failure_count_ = 0;
|
|
}
|
|
|
|
int wait_append_sync(const share::ObLSID &ls_id)
|
|
{
|
|
WaitQuiescent(ls_qs_);
|
|
CLOG_LOG(INFO, "WaitQuiescent", K(ls_id));
|
|
return OB_SUCCESS;
|
|
}
|
|
void critical_guard()
|
|
{
|
|
CriticalGuard(ls_qs_);
|
|
}
|
|
void inc_success_count()
|
|
{
|
|
ATOMIC_INC(&success_count_);
|
|
}
|
|
void inc_failure_count()
|
|
{
|
|
ATOMIC_INC(&failure_count_);
|
|
}
|
|
int replay(ObLogReplayTask *replay_task){
|
|
UNUSED(replay_task);
|
|
return OB_SUCCESS;
|
|
}
|
|
int64_t success_count_;
|
|
int64_t failure_count_;
|
|
common::ObQSync ls_qs_;
|
|
};
|
|
|
|
class TestObSimpleLogApplyFunc : public ObSimpleLogClusterTestEnv
|
|
{
|
|
public:
|
|
TestObSimpleLogApplyFunc() : ObSimpleLogClusterTestEnv()
|
|
{}
|
|
};
|
|
|
|
class MockAppendCb : public AppendCb
|
|
{
|
|
public:
|
|
int on_success()
|
|
{
|
|
ls_adapter_->inc_success_count();
|
|
CLOG_LOG(INFO, "on_success", K(log_id_));
|
|
return OB_SUCCESS;
|
|
}
|
|
|
|
int on_failure()
|
|
{
|
|
ls_adapter_->inc_failure_count();
|
|
CLOG_LOG(INFO, "on_failure", K(log_id_));
|
|
return OB_SUCCESS;
|
|
}
|
|
|
|
void init(const int64_t log_id,
|
|
MockLSAdapter *ls_adapter)
|
|
{
|
|
log_id_ = log_id;
|
|
ls_adapter_ = ls_adapter;
|
|
}
|
|
int64_t log_id_;
|
|
MockLSAdapter *ls_adapter_;
|
|
};
|
|
|
|
int64_t ObSimpleLogClusterTestBase::member_cnt_ = 3;
|
|
int64_t ObSimpleLogClusterTestBase::node_cnt_ = 3;
|
|
std::string ObSimpleLogClusterTestBase::test_name_ = TEST_NAME;
|
|
bool ObSimpleLogClusterTestBase::need_add_arb_server_ = false;
|
|
bool ObSimpleLogClusterTestBase::need_shared_storage_ = false;
|
|
|
|
TEST_F(TestObSimpleLogApplyFunc, apply)
|
|
{
|
|
const int64_t task_count = 50;
|
|
const int64_t truncate_count = 25;
|
|
const int64_t id = ATOMIC_AAF(&palf_id_, 1);
|
|
ObLSID ls_id(id);
|
|
int64_t leader_idx = 0;
|
|
LSN basic_lsn(0);
|
|
PalfHandleImplGuard leader;
|
|
MockAppendCb *cb_array[task_count];
|
|
LSN unused_apply_end_lsn;
|
|
bool is_apply_done = false;
|
|
CLOG_LOG(INFO, "test apply begin", K(id));
|
|
EXPECT_EQ(OB_SUCCESS, create_paxos_group(id, leader_idx, leader));
|
|
MockLSAdapter ls_adapter;
|
|
ls_adapter.init((ObLSService *)(0x1));
|
|
ObLogApplyService ap_sv;
|
|
for (int i = 0; i < task_count; i++)
|
|
{
|
|
cb_array[i] = new MockAppendCb();
|
|
cb_array[i]->init(i + 1, &ls_adapter);
|
|
}
|
|
PalfEnv *palf_env;
|
|
EXPECT_EQ(OB_SUCCESS, get_palf_env(leader_idx, palf_env));
|
|
EXPECT_EQ(OB_SUCCESS, ap_sv.init(palf_env, &ls_adapter));
|
|
EXPECT_EQ(OB_SUCCESS, ap_sv.start());
|
|
EXPECT_EQ(OB_SUCCESS, ap_sv.add_ls(ls_id));
|
|
EXPECT_EQ(OB_SUCCESS, ap_sv.switch_to_leader(ls_id, 1));
|
|
const int64_t idx_1 = (leader_idx + 1) % get_node_cnt();
|
|
const int64_t idx_2 = (leader_idx + 2) % get_node_cnt();
|
|
|
|
//旧主少数派写日志
|
|
block_net(leader_idx, idx_1);
|
|
block_net(leader_idx, idx_2);
|
|
do {
|
|
ObApplyStatus *apply_status = NULL;
|
|
ObApplyStatusGuard guard;
|
|
ap_sv.get_apply_status(ls_id, guard);
|
|
apply_status = guard.get_apply_status();
|
|
LSN lsn;
|
|
share::SCN scn;
|
|
int ret = OB_SUCCESS;
|
|
for (int i = 0; i < truncate_count; i++)
|
|
{
|
|
{
|
|
ls_adapter.critical_guard();
|
|
CLOG_LOG(INFO, "submit log start", K(i));
|
|
share::SCN ref_scn;
|
|
ref_scn.convert_for_logservice(i);
|
|
ret = submit_log(leader, ref_scn, lsn, scn);
|
|
if (OB_SUCC(ret)) {
|
|
EXPECT_EQ(true, lsn.is_valid());
|
|
cb_array[i]->__set_lsn(lsn);
|
|
cb_array[i]->__set_scn(scn);
|
|
EXPECT_EQ(OB_SUCCESS, apply_status->push_append_cb(cb_array[i]));
|
|
} else {
|
|
break;
|
|
}
|
|
}
|
|
CLOG_LOG(INFO, "submit log finish", K(i), K(lsn), K(scn));
|
|
}
|
|
} while (0);
|
|
share::SCN min_scn;
|
|
EXPECT_EQ(OB_SUCCESS, ap_sv.get_max_applied_scn(ls_id, min_scn));
|
|
EXPECT_EQ(OB_SUCCESS, ap_sv.switch_to_follower(ls_id));
|
|
|
|
//切主, truncate旧主日志,预期所有cb都调用on_failure
|
|
sleep(15);
|
|
while (!is_apply_done)
|
|
{
|
|
ap_sv.is_apply_done(ls_id, is_apply_done, unused_apply_end_lsn);
|
|
usleep(100);
|
|
}
|
|
|
|
//切回旧主写日志,预期所有cb都调用on_success
|
|
unblock_net(leader_idx, idx_1);
|
|
unblock_net(leader_idx, idx_2);
|
|
int64_t new_leader_idx = 0;
|
|
PalfHandleImplGuard new_leader;
|
|
EXPECT_EQ(OB_SUCCESS, get_leader(id, new_leader, new_leader_idx));
|
|
EXPECT_NE(new_leader_idx, leader_idx);
|
|
//等待membership同步
|
|
sleep(2);
|
|
leader.reset();
|
|
CLOG_LOG(INFO, "new leader", K(new_leader_idx), K(leader_idx));
|
|
EXPECT_EQ(OB_SUCCESS, switch_leader(id, leader_idx, leader));
|
|
leader.reset();
|
|
sleep(2);
|
|
EXPECT_EQ(OB_SUCCESS, ap_sv.switch_to_leader(ls_id, 3));
|
|
EXPECT_EQ(OB_SUCCESS, ap_sv.get_max_applied_scn(ls_id, min_scn));
|
|
EXPECT_EQ(OB_SUCCESS, get_leader(id, leader, new_leader_idx));
|
|
CLOG_LOG(INFO, "check switch leader", K(new_leader_idx), K(leader_idx));
|
|
EXPECT_EQ(new_leader_idx, leader_idx);
|
|
share::SCN palf_end_scn;
|
|
do {
|
|
ObApplyStatus *apply_status = NULL;
|
|
ObApplyStatusGuard guard;
|
|
ap_sv.get_apply_status(ls_id, guard);
|
|
apply_status = guard.get_apply_status();
|
|
LSN lsn;
|
|
share::SCN scn;
|
|
for (int i = truncate_count; i < task_count; i++)
|
|
{
|
|
EXPECT_EQ(OB_SUCCESS, ap_sv.get_max_applied_scn(ls_id, scn));
|
|
{
|
|
int64_t ref_ts = i - truncate_count;
|
|
share::SCN ref_scn;
|
|
ref_scn.convert_for_logservice(ref_ts);
|
|
ls_adapter.critical_guard();
|
|
CLOG_LOG(INFO, "submit log start", K(i));
|
|
EXPECT_EQ(OB_SUCCESS, submit_log(leader, ref_scn, lsn, scn));
|
|
EXPECT_EQ(true, lsn.is_valid());
|
|
cb_array[i]->__set_lsn(lsn);
|
|
cb_array[i]->__set_scn(scn);
|
|
palf_end_scn = scn;
|
|
EXPECT_EQ(OB_SUCCESS, apply_status->push_append_cb(cb_array[i]));
|
|
}
|
|
CLOG_LOG(INFO, "submit log finish", K(i), K(lsn), K(scn));
|
|
EXPECT_EQ(OB_SUCCESS, ap_sv.get_max_applied_scn(ls_id, scn));
|
|
}
|
|
} while (0);
|
|
CLOG_LOG(INFO, "truncate write finish", K(id));
|
|
|
|
// remove
|
|
is_apply_done = false;
|
|
while (!is_apply_done)
|
|
{
|
|
ap_sv.is_apply_done(ls_id, is_apply_done, unused_apply_end_lsn);
|
|
usleep(100);
|
|
}
|
|
share::SCN scn;
|
|
EXPECT_EQ(OB_SUCCESS, ap_sv.switch_to_follower(ls_id));
|
|
EXPECT_EQ(OB_SUCCESS, ap_sv.get_max_applied_scn(ls_id, scn));
|
|
EXPECT_EQ(scn.get_val_for_logservice(), palf_end_scn.get_val_for_logservice());
|
|
EXPECT_EQ(truncate_count, ls_adapter.failure_count_);
|
|
EXPECT_EQ(task_count- truncate_count, ls_adapter.success_count_);
|
|
for (int i = 0; i < task_count; i++)
|
|
{
|
|
delete cb_array[i];
|
|
}
|
|
EXPECT_EQ(OB_SUCCESS, ap_sv.remove_ls(ls_id));
|
|
ap_sv.stop();
|
|
ap_sv.wait();
|
|
ap_sv.destroy();
|
|
CLOG_LOG(INFO, "test apply finish", K(id));
|
|
}
|
|
|
|
TEST_F(TestObSimpleLogApplyFunc, get_max_decided_scn)
|
|
{
|
|
const int64_t task_count = 10;
|
|
const int64_t id = ATOMIC_AAF(&palf_id_, 1);
|
|
ObLSID ls_id(id);
|
|
int64_t leader_idx = 0;
|
|
LSN basic_lsn(0);
|
|
PalfHandleImplGuard leader;
|
|
MockAppendCb *cb_array[task_count];
|
|
CLOG_LOG(INFO, "test get_max_decided_scn begin", K(id));
|
|
EXPECT_EQ(OB_SUCCESS, create_paxos_group(id, leader_idx, leader));
|
|
MockLSAdapter ls_adapter;
|
|
ls_adapter.init((ObLSService *)(0x1));
|
|
ObLogApplyService ap_sv;
|
|
//submit log first
|
|
for (int i = 0; i < task_count; i++)
|
|
{
|
|
cb_array[i] = new MockAppendCb();
|
|
cb_array[i]->init(i + 1, &ls_adapter);
|
|
}
|
|
PalfEnv *palf_env;
|
|
ObTenantEnv::set_tenant(get_cluster()[leader_idx]->get_tenant_base());
|
|
EXPECT_EQ(OB_SUCCESS, get_palf_env(leader_idx, palf_env));
|
|
EXPECT_EQ(OB_SUCCESS, ap_sv.init(palf_env, &ls_adapter));
|
|
EXPECT_EQ(OB_SUCCESS, ap_sv.start());
|
|
EXPECT_EQ(OB_SUCCESS, ap_sv.add_ls(ls_id));
|
|
EXPECT_EQ(OB_SUCCESS, ap_sv.switch_to_leader(ls_id, 1));
|
|
const int64_t idx_1 = (leader_idx + 1) % get_node_cnt();
|
|
const int64_t idx_2 = (leader_idx + 2) % get_node_cnt();
|
|
share::SCN ref_scn;
|
|
int i = 0;
|
|
LSN lsn;
|
|
share::SCN scn;
|
|
|
|
do {
|
|
ObApplyStatus *apply_status = NULL;
|
|
ObApplyStatusGuard guard;
|
|
ap_sv.get_apply_status(ls_id, guard);
|
|
apply_status = guard.get_apply_status();
|
|
ASSERT_TRUE(NULL != apply_status);
|
|
int ret = OB_SUCCESS;
|
|
for (i = 0; i < task_count; i++)
|
|
{
|
|
{
|
|
ls_adapter.critical_guard();
|
|
CLOG_LOG(INFO, "submit log start", K(i));
|
|
ref_scn.convert_for_logservice(i);
|
|
ret = submit_log(leader, ref_scn, lsn, scn);
|
|
if (OB_SUCC(ret)) {
|
|
EXPECT_EQ(true, lsn.is_valid());
|
|
cb_array[i]->__set_lsn(lsn);
|
|
cb_array[i]->__set_scn(scn);
|
|
EXPECT_EQ(OB_SUCCESS, apply_status->push_append_cb(cb_array[i]));
|
|
} else {
|
|
break;
|
|
}
|
|
}
|
|
CLOG_LOG(INFO, "submit log finish", K(i), K(lsn), K(scn));
|
|
}
|
|
} while (0);
|
|
|
|
LSN max_lsn = leader.palf_handle_impl_->sw_.get_max_lsn();
|
|
SCN max_scn = leader.palf_handle_impl_->sw_.get_max_scn();
|
|
|
|
//wait apply done
|
|
wait_lsn_until_slide(max_lsn, leader);
|
|
EXPECT_EQ(OB_SUCCESS, ap_sv.switch_to_follower(ls_id));
|
|
LSN apply_end_lsn;
|
|
bool is_apply_done = false;
|
|
while (!is_apply_done)
|
|
{
|
|
ap_sv.is_apply_done(ls_id, is_apply_done, apply_end_lsn);
|
|
usleep(100);
|
|
}
|
|
EXPECT_EQ(max_lsn, apply_end_lsn);
|
|
|
|
//switch leader and submit log
|
|
PalfHandleImplGuard new_leader;
|
|
EXPECT_EQ(OB_SUCCESS, switch_leader(id, idx_1, new_leader));
|
|
|
|
PalfHandleImplGuard new_get_leader;
|
|
int64_t new_leader_idx = 0;
|
|
EXPECT_EQ(OB_SUCCESS, get_leader(id, new_get_leader, new_leader_idx));
|
|
EXPECT_NE(new_leader_idx, leader_idx);
|
|
|
|
int ret = OB_SUCCESS;
|
|
++i;
|
|
ref_scn.convert_for_logservice(i);
|
|
ret = submit_log(new_leader, ref_scn, lsn, scn);
|
|
EXPECT_EQ(OB_SUCCESS, ret);
|
|
++i;
|
|
ref_scn.convert_for_logservice(i);
|
|
ret = submit_log(new_leader, ref_scn, lsn, scn);
|
|
EXPECT_EQ(OB_SUCCESS, ret);
|
|
|
|
LSN end_lsn;
|
|
do {
|
|
end_lsn = leader.palf_handle_impl_->get_end_lsn();
|
|
if (end_lsn <= max_lsn) {
|
|
CLOG_LOG(INFO, "wait end_lsn", K(end_lsn), K(max_lsn));
|
|
ret = submit_log(new_leader, 1, id, 100);
|
|
wait_until_has_committed(new_leader, new_leader.palf_handle_impl_->get_max_lsn());
|
|
sleep (1);
|
|
}
|
|
} while (end_lsn <= max_lsn);
|
|
share::SCN max_decided_scn;
|
|
EXPECT_EQ(OB_SUCCESS, ap_sv.get_max_applied_scn(ls_id, max_decided_scn));
|
|
CLOG_LOG(INFO, "check here", K(max_scn), K(max_decided_scn), K(end_lsn), K(max_lsn), K(apply_end_lsn));
|
|
EXPECT_EQ(max_decided_scn, max_scn);
|
|
|
|
for (int i = 0; i < task_count; i++)
|
|
{
|
|
delete cb_array[i];
|
|
}
|
|
|
|
ObTenantEnv::set_tenant(get_cluster()[leader_idx]->get_tenant_base());
|
|
EXPECT_EQ(OB_SUCCESS, ap_sv.remove_ls(ls_id));
|
|
ap_sv.stop();
|
|
ap_sv.wait();
|
|
ap_sv.destroy();
|
|
CLOG_LOG(INFO, "test get_max_decided_scn finish", K(id));
|
|
}
|
|
|
|
TEST_F(TestObSimpleLogApplyFunc, get_max_decided_scn_no_more_log_after_switch_to_follower)
|
|
{
|
|
const int64_t task_count = 10;
|
|
const int64_t id = ATOMIC_AAF(&palf_id_, 1);
|
|
ObLSID ls_id(id);
|
|
int64_t leader_idx = 0;
|
|
LSN basic_lsn(0);
|
|
PalfHandleImplGuard leader;
|
|
MockAppendCb *cb_array[task_count];
|
|
CLOG_LOG(INFO, "test get_max_decided_scn_no_more_log_after_switch_to_follower begin", K(id));
|
|
EXPECT_EQ(OB_SUCCESS, create_paxos_group(id, leader_idx, leader));
|
|
MockLSAdapter ls_adapter;
|
|
ls_adapter.init((ObLSService *)(0x1));
|
|
ObLogApplyService ap_sv;
|
|
//submit log first
|
|
for (int i = 0; i < task_count; i++)
|
|
{
|
|
cb_array[i] = new MockAppendCb();
|
|
cb_array[i]->init(i + 1, &ls_adapter);
|
|
}
|
|
PalfEnv *palf_env;
|
|
ObTenantEnv::set_tenant(get_cluster()[leader_idx]->get_tenant_base());
|
|
EXPECT_EQ(OB_SUCCESS, get_palf_env(leader_idx, palf_env));
|
|
EXPECT_EQ(OB_SUCCESS, ap_sv.init(palf_env, &ls_adapter));
|
|
EXPECT_EQ(OB_SUCCESS, ap_sv.start());
|
|
EXPECT_EQ(OB_SUCCESS, ap_sv.add_ls(ls_id));
|
|
EXPECT_EQ(OB_SUCCESS, ap_sv.switch_to_leader(ls_id, 1));
|
|
const int64_t idx_1 = (leader_idx + 1) % get_node_cnt();
|
|
const int64_t idx_2 = (leader_idx + 2) % get_node_cnt();
|
|
share::SCN ref_scn;
|
|
int i = 0;
|
|
LSN lsn;
|
|
share::SCN scn;
|
|
|
|
do {
|
|
ObApplyStatus *apply_status = NULL;
|
|
ObApplyStatusGuard guard;
|
|
ap_sv.get_apply_status(ls_id, guard);
|
|
apply_status = guard.get_apply_status();
|
|
ASSERT_TRUE(NULL != apply_status);
|
|
int ret = OB_SUCCESS;
|
|
for (i = 0; i < task_count; i++)
|
|
{
|
|
{
|
|
ls_adapter.critical_guard();
|
|
CLOG_LOG(INFO, "submit log start", K(i));
|
|
ref_scn.convert_for_logservice(i);
|
|
ret = submit_log(leader, ref_scn, lsn, scn);
|
|
if (OB_SUCC(ret)) {
|
|
EXPECT_EQ(true, lsn.is_valid());
|
|
cb_array[i]->__set_lsn(lsn);
|
|
cb_array[i]->__set_scn(scn);
|
|
EXPECT_EQ(OB_SUCCESS, apply_status->push_append_cb(cb_array[i]));
|
|
} else {
|
|
break;
|
|
}
|
|
}
|
|
CLOG_LOG(INFO, "submit log finish", K(i), K(lsn), K(scn));
|
|
}
|
|
} while (0);
|
|
|
|
LSN max_lsn = leader.palf_handle_impl_->sw_.get_max_lsn();
|
|
SCN max_scn = leader.palf_handle_impl_->sw_.get_max_scn();
|
|
|
|
//wait apply done
|
|
wait_lsn_until_slide(max_lsn, leader);
|
|
EXPECT_EQ(OB_SUCCESS, ap_sv.switch_to_follower(ls_id));
|
|
LSN apply_end_lsn;
|
|
bool is_apply_done = false;
|
|
while (!is_apply_done)
|
|
{
|
|
ap_sv.is_apply_done(ls_id, is_apply_done, apply_end_lsn);
|
|
usleep(100);
|
|
}
|
|
EXPECT_EQ(max_lsn, apply_end_lsn);
|
|
|
|
share::SCN max_decided_scn;
|
|
EXPECT_EQ(OB_SUCCESS, ap_sv.get_max_applied_scn(ls_id, max_decided_scn));
|
|
CLOG_LOG(INFO, "check here", K(max_scn), K(max_decided_scn), K(max_lsn), K(apply_end_lsn));
|
|
EXPECT_EQ(max_decided_scn, max_scn);
|
|
|
|
ObTenantEnv::set_tenant(get_cluster()[leader_idx]->get_tenant_base());
|
|
EXPECT_EQ(OB_SUCCESS, ap_sv.remove_ls(ls_id));
|
|
ap_sv.stop();
|
|
ap_sv.wait();
|
|
ap_sv.destroy();
|
|
CLOG_LOG(INFO, "test get_max_decided_scn_with_no_more_log finish", K(id));
|
|
}
|
|
|
|
} // unitest
|
|
} // oceanbase
|
|
|
|
int main(int argc, char **argv)
|
|
{
|
|
RUN_SIMPLE_LOG_CLUSTER_TEST(TEST_NAME);
|
|
}
|