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

fix typos from open github issue

Browse Source
This commit is contained in:
xy0
2022-03-04 19:48:51 +08:00
committed by LINxiansheng
parent 42fcd3dd80
commit 25bd25175f
4 changed files with 1320 additions and 1325 deletions
Download Patch File Download Diff File Expand all files Collapse all files

180
src/election/ob_election.cpp
View File

@ -36,7 +36,7 @@ namespace election {
/**********should removed after a barrier version bigger than 3.1**********/ /**********should removed after a barrier version bigger than 3.1**********/
bool ObElection::IS_CLUSTER_MIN_VSERDION_LESS_THAN_3_1 = true; bool ObElection::IS_CLUSTER_MIN_VSERDION_LESS_THAN_3_1 = true;
/**************************************************************************************************/ /**************************************************************************************************/
const char* const ObElection::REVOKE_REASON_STR[REVOKE_TYPE_MAX] = {"leader lease expired", const char *const ObElection::REVOKE_REASON_STR[REVOKE_TYPE_MAX] = {"leader lease expired",
"self is not candidate", "self is not candidate",
"disk error", "disk error",
"clog reconfirm timeout", "clog reconfirm timeout",
@ -89,9 +89,9 @@ void ObElection::reset()
cached_lease_end_ = 0; cached_lease_end_ = 0;
} }
int ObElection::init(const ObPartitionKey& partition, const ObAddr& self, ObIElectionRpc* rpc, ObTimeWheel* tw, int ObElection::init(const ObPartitionKey &partition, const ObAddr &self, ObIElectionRpc *rpc, ObTimeWheel *tw,
const int64_t replica_num, ObIElectionCallback* election_cb, ObIElectionGroupMgr* eg_mgr, const int64_t replica_num, ObIElectionCallback *election_cb, ObIElectionGroupMgr *eg_mgr,
ObElectionEventHistoryArray* event_history_array) ObElectionEventHistoryArray *event_history_array)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -289,7 +289,7 @@ int ObElection::stop()
return ret; return ret;
} }
int ObElection::process_devote_prepare_(const ObElectionMsgDEPrepare& msg) int ObElection::process_devote_prepare_(const ObElectionMsgDEPrepare &msg)
{ {
// requires caller to lock and check // requires caller to lock and check
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -335,7 +335,7 @@ int ObElection::process_devote_prepare_(const ObElectionMsgDEPrepare& msg)
return ret; return ret;
} }
int ObElection::local_handle_devote_prepare_(const ObElectionMsgDEPrepare& msg) int ObElection::local_handle_devote_prepare_(const ObElectionMsgDEPrepare &msg)
{ {
// requires caller to lock and check // requires caller to lock and check
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -370,7 +370,7 @@ int ObElection::local_handle_devote_prepare_(const ObElectionMsgDEPrepare& msg)
return ret; return ret;
} }
int ObElection::handle_devote_prepare(const ObElectionMsgDEPrepare& msg, ObElectionRpcResult& result) int ObElection::handle_devote_prepare(const ObElectionMsgDEPrepare &msg, ObElectionRpcResult &result)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
const int64_t cur_ts = get_current_ts(); const int64_t cur_ts = get_current_ts();
@ -394,7 +394,7 @@ int ObElection::handle_devote_prepare(const ObElectionMsgDEPrepare& msg, ObElect
} else if (current_leader_.is_valid() && OB_SUCCESS == verify_leader_valid_(cur_ts)) { } else if (current_leader_.is_valid() && OB_SUCCESS == verify_leader_valid_(cur_ts)) {
// if current leader is valid, some one ask for decentralized voting, meaning the requester don't know the leader // if current leader is valid, some one ask for decentralized voting, meaning the requester don't know the leader
// notify him who is leader // notify him who is leader
const ObAddr& sender = msg.get_sender(); const ObAddr &sender = msg.get_sender();
if (current_leader_ != sender && self_ != sender) { if (current_leader_ != sender && self_ != sender) {
if (OB_FAIL(notify_leader_(sender))) { if (OB_FAIL(notify_leader_(sender))) {
ELECT_ASYNC_LOG_(WARN, "notify leader error", K(ret), "election", *this, K(msg)); ELECT_ASYNC_LOG_(WARN, "notify leader error", K(ret), "election", *this, K(msg));
@ -429,7 +429,7 @@ int ObElection::handle_devote_prepare(const ObElectionMsgDEPrepare& msg, ObElect
return ret; return ret;
} }
int ObElection::process_devote_vote_(const ObElectionMsgDEVote& msg) int ObElection::process_devote_vote_(const ObElectionMsgDEVote &msg)
{ {
// requires caller to lock and check // requires caller to lock and check
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -469,7 +469,7 @@ int ObElection::process_devote_vote_(const ObElectionMsgDEVote& msg)
return ret; return ret;
} }
int ObElection::local_handle_devote_vote_(const ObElectionMsgDEVote& msg) int ObElection::local_handle_devote_vote_(const ObElectionMsgDEVote &msg)
{ {
// requires caller to lock and check // requires caller to lock and check
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -500,7 +500,7 @@ int ObElection::local_handle_devote_vote_(const ObElectionMsgDEVote& msg)
return ret; return ret;
} }
int ObElection::handle_devote_vote(const ObElectionMsgDEVote& msg, ObElectionRpcResult& result) int ObElection::handle_devote_vote(const ObElectionMsgDEVote &msg, ObElectionRpcResult &result)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -538,7 +538,7 @@ int ObElection::handle_devote_vote(const ObElectionMsgDEVote& msg, ObElectionRpc
return ret; return ret;
} }
int ObElection::process_devote_success_(const ObElectionMsgDESuccess& msg) int ObElection::process_devote_success_(const ObElectionMsgDESuccess &msg)
{ {
// requires caller to lock and check // requires caller to lock and check
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -554,7 +554,7 @@ int ObElection::process_devote_success_(const ObElectionMsgDESuccess& msg)
ELECT_ASYNC_LOG_(WARN, "invalid argument", K(ret), K_(partition), K(msg)); ELECT_ASYNC_LOG_(WARN, "invalid argument", K(ret), K_(partition), K(msg));
} else { } else {
const int64_t cur_ts = get_current_ts(); const int64_t cur_ts = get_current_ts();
const ObAddr& msg_leader = msg.get_leader(); const ObAddr &msg_leader = msg.get_leader();
const int64_t msg_t1 = msg.get_T1_timestamp(); const int64_t msg_t1 = msg.get_T1_timestamp();
const int64_t msg_real_t1 = get_msg_real_T1_(msg); const int64_t msg_real_t1 = get_msg_real_T1_(msg);
const int64_t lease_time = msg.get_lease_time(); const int64_t lease_time = msg.get_lease_time();
@ -582,7 +582,7 @@ int ObElection::process_devote_success_(const ObElectionMsgDESuccess& msg)
(void)update_info_from_eg_(); (void)update_info_from_eg_();
} }
ret = OB_ELECTION_WARN_PROTOCOL_ERROR; ret = OB_ELECTION_WARN_PROTOCOL_ERROR;
FORCE_ELECT_LOG(WARN, "leader available but recevive devote_success message", K(ret), K(*this), K(msg)); FORCE_ELECT_LOG(WARN, "leader available but receive devote_success message", K(ret), K(*this), K(msg));
} }
} else if (NULL != election_group_) { } else if (NULL != election_group_) {
// current leader is invalid, need move out from election group // current leader is invalid, need move out from election group
@ -618,7 +618,7 @@ int ObElection::process_devote_success_(const ObElectionMsgDESuccess& msg)
return ret; return ret;
} }
int ObElection::local_handle_devote_success_(const ObElectionMsgDESuccess& msg) int ObElection::local_handle_devote_success_(const ObElectionMsgDESuccess &msg)
{ {
// requires caller to lock and check // requires caller to lock and check
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -633,7 +633,7 @@ int ObElection::local_handle_devote_success_(const ObElectionMsgDESuccess& msg)
ret = OB_INVALID_ARGUMENT; ret = OB_INVALID_ARGUMENT;
ELECT_ASYNC_LOG_(WARN, "invalid argument", K(ret), K_(partition), K(msg)); ELECT_ASYNC_LOG_(WARN, "invalid argument", K(ret), K_(partition), K(msg));
} else if (OB_FAIL(process_devote_success_(msg))) { } else if (OB_FAIL(process_devote_success_(msg))) {
ELECT_ASYNC_LOG_(WARN, "process_devote_success_ faield", K(ret), K_(partition), K(msg)); ELECT_ASYNC_LOG_(WARN, "process_devote_success_ failed", K(ret), K_(partition), K(msg));
} else { } else {
} }
@ -645,7 +645,7 @@ int ObElection::local_handle_devote_success_(const ObElectionMsgDESuccess& msg)
return ret; return ret;
} }
int ObElection::handle_devote_success(const ObElectionMsgDESuccess& msg, ObElectionRpcResult& result) int ObElection::handle_devote_success(const ObElectionMsgDESuccess &msg, ObElectionRpcResult &result)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -679,7 +679,7 @@ int ObElection::handle_devote_success(const ObElectionMsgDESuccess& msg, ObElect
return ret; return ret;
} }
int ObElection::process_vote_prepare_(const ObElectionMsgPrepare& msg) int ObElection::process_vote_prepare_(const ObElectionMsgPrepare &msg)
{ {
// requires caller to lock and check // requires caller to lock and check
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -770,7 +770,7 @@ int ObElection::process_vote_prepare_(const ObElectionMsgPrepare& msg)
return ret; return ret;
} }
int ObElection::local_handle_vote_prepare_(const ObElectionMsgPrepare& msg) int ObElection::local_handle_vote_prepare_(const ObElectionMsgPrepare &msg)
{ {
// requires caller to lock and check // requires caller to lock and check
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -799,7 +799,7 @@ int ObElection::local_handle_vote_prepare_(const ObElectionMsgPrepare& msg)
return ret; return ret;
} }
int ObElection::handle_vote_prepare(const ObElectionMsgPrepare& msg, ObElectionRpcResult& result) int ObElection::handle_vote_prepare(const ObElectionMsgPrepare &msg, ObElectionRpcResult &result)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -838,7 +838,7 @@ int ObElection::handle_vote_prepare(const ObElectionMsgPrepare& msg, ObElectionR
return ret; return ret;
} }
int ObElection::process_vote_vote_(const ObElectionMsgVote& msg) int ObElection::process_vote_vote_(const ObElectionMsgVote &msg)
{ {
// requires caller to lock and check // requires caller to lock and check
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -908,7 +908,7 @@ int ObElection::process_vote_vote_(const ObElectionMsgVote& msg)
return ret; return ret;
} }
int ObElection::local_handle_vote_vote_(const ObElectionMsgVote& msg) int ObElection::local_handle_vote_vote_(const ObElectionMsgVote &msg)
{ {
// requires caller to lock and check // requires caller to lock and check
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -941,7 +941,7 @@ int ObElection::local_handle_vote_vote_(const ObElectionMsgVote& msg)
return ret; return ret;
} }
int ObElection::handle_vote_vote(const ObElectionMsgVote& msg, ObElectionRpcResult& result) int ObElection::handle_vote_vote(const ObElectionMsgVote &msg, ObElectionRpcResult &result)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -981,13 +981,13 @@ int ObElection::handle_vote_vote(const ObElectionMsgVote& msg, ObElectionRpcResu
return ret; return ret;
} }
int ObElection::process_vote_success_(const ObElectionMsgSuccess& msg) int ObElection::process_vote_success_(const ObElectionMsgSuccess &msg)
{ {
// requires caller to lock and check // requires caller to lock and check
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
if (msg.get_last_leader_epoch() != OB_INVALID_TIMESTAMP && if (msg.get_last_leader_epoch() != OB_INVALID_TIMESTAMP &&
(max_leader_epoch_ever_seen_ == OB_INVALID_TIMESTAMP || // avoid OB_INVALID_TIMESTAMP definded too large (max_leader_epoch_ever_seen_ == OB_INVALID_TIMESTAMP || // avoid OB_INVALID_TIMESTAMP defined too large
msg.get_last_leader_epoch() > max_leader_epoch_ever_seen_)) { msg.get_last_leader_epoch() > max_leader_epoch_ever_seen_)) {
max_leader_epoch_ever_seen_ = msg.get_last_leader_epoch(); max_leader_epoch_ever_seen_ = msg.get_last_leader_epoch();
} }
@ -1011,8 +1011,8 @@ int ObElection::process_vote_success_(const ObElectionMsgSuccess& msg)
const int64_t lease_time = msg.get_lease_time(); const int64_t lease_time = msg.get_lease_time();
const int64_t now_lease_time = (lease_time == 0) ? (OB_ELECTION_130_LEASE_TIME) : lease_time; const int64_t now_lease_time = (lease_time == 0) ? (OB_ELECTION_130_LEASE_TIME) : lease_time;
const ObAddr& msg_new_leader = msg.get_new_leader(); const ObAddr &msg_new_leader = msg.get_new_leader();
const ObAddr& msg_cur_leader = msg.get_cur_leader(); const ObAddr &msg_cur_leader = msg.get_cur_leader();
ObAddr cur_leader; ObAddr cur_leader;
if (!IN_T0_RANGE(cur_ts, msg_real_t1)) { if (!IN_T0_RANGE(cur_ts, msg_real_t1)) {
@ -1133,7 +1133,7 @@ int ObElection::process_vote_success_(const ObElectionMsgSuccess& msg)
return ret; return ret;
} }
int ObElection::local_handle_vote_success_(const ObElectionMsgSuccess& msg) int ObElection::local_handle_vote_success_(const ObElectionMsgSuccess &msg)
{ {
// requires caller to lock and check // requires caller to lock and check
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -1162,7 +1162,7 @@ int ObElection::local_handle_vote_success_(const ObElectionMsgSuccess& msg)
return ret; return ret;
} }
int ObElection::handle_vote_success(const ObElectionMsgSuccess& msg, ObElectionRpcResult& result) int ObElection::handle_vote_success(const ObElectionMsgSuccess &msg, ObElectionRpcResult &result)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -1198,7 +1198,7 @@ int ObElection::handle_vote_success(const ObElectionMsgSuccess& msg, ObElectionR
return ret; return ret;
} }
int ObElection::handle_query_leader(const ObElectionQueryLeader& msg, ObElectionRpcResult& result) int ObElection::handle_query_leader(const ObElectionQueryLeader &msg, ObElectionRpcResult &result)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -1215,10 +1215,10 @@ int ObElection::handle_query_leader(const ObElectionQueryLeader& msg, ObElection
ELECT_ASYNC_LOG_(WARN, "election is not running", K(ret), K_(partition), K(msg)); ELECT_ASYNC_LOG_(WARN, "election is not running", K(ret), K_(partition), K(msg));
} else if (!msg.is_valid()) { } else if (!msg.is_valid()) {
ret = OB_INVALID_ARGUMENT; ret = OB_INVALID_ARGUMENT;
ELECT_ASYNC_LOG_(WARN, "invalid argumenet", K(ret), K_(partition), K(msg)); ELECT_ASYNC_LOG_(WARN, "invalid argument", K(ret), K_(partition), K(msg));
} else { } else {
const int64_t cur_ts = get_current_ts(); const int64_t cur_ts = get_current_ts();
const ObAddr& sender = msg.get_sender(); const ObAddr &sender = msg.get_sender();
// defense // defense
if (OB_SUCC(verify_leader_valid_(cur_ts)) // leader must be valid if (OB_SUCC(verify_leader_valid_(cur_ts)) // leader must be valid
&& current_leader_ != sender // sender can't be leader && current_leader_ != sender // sender can't be leader
@ -1236,7 +1236,7 @@ int ObElection::handle_query_leader(const ObElectionQueryLeader& msg, ObElection
return ret; return ret;
} }
int ObElection::handle_query_leader_response(const ObElectionQueryLeaderResponse& msg, ObElectionRpcResult& result) int ObElection::handle_query_leader_response(const ObElectionQueryLeaderResponse &msg, ObElectionRpcResult &result)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -1258,7 +1258,7 @@ int ObElection::handle_query_leader_response(const ObElectionQueryLeaderResponse
ret = OB_STATE_NOT_MATCH; ret = OB_STATE_NOT_MATCH;
ELECT_ASYNC_LOG_(WARN, "already in election_group, ignore msg", K(ret), "election", *this, K(msg)); ELECT_ASYNC_LOG_(WARN, "already in election_group, ignore msg", K(ret), "election", *this, K(msg));
} else { } else {
const ObAddr& msg_leader = msg.get_leader(); const ObAddr &msg_leader = msg.get_leader();
const int64_t t1 = msg.get_t1(); const int64_t t1 = msg.get_t1();
const int64_t epoch = msg.get_epoch(); const int64_t epoch = msg.get_epoch();
const int64_t lease_time = msg.get_lease_time(); const int64_t lease_time = msg.get_lease_time();
@ -1313,7 +1313,7 @@ int ObElection::handle_query_leader_response(const ObElectionQueryLeaderResponse
} }
int ObElection::set_candidate( int ObElection::set_candidate(
const int64_t replica_num, const ObMemberList& curr_mlist, const int64_t membership_version) const int64_t replica_num, const ObMemberList &curr_mlist, const int64_t membership_version)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -1397,7 +1397,7 @@ int ObElection::force_leader_async()
return ret; return ret;
} }
int ObElection::change_leader_async(const ObAddr& leader, ObTsWindows& changing_leader_windows) int ObElection::change_leader_async(const ObAddr &leader, ObTsWindows &changing_leader_windows)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -1451,7 +1451,7 @@ int ObElection::change_leader_to_self()
return ret; return ret;
} }
int ObElection::get_curr_candidate(ObMemberList& mlist) const int ObElection::get_curr_candidate(ObMemberList &mlist) const
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -1470,7 +1470,7 @@ int ObElection::get_curr_candidate(ObMemberList& mlist) const
// only leader could call this interface // only leader could call this interface
// return OB_NOT_MASTER if caller is follower // return OB_NOT_MASTER if caller is follower
int ObElection::get_valid_candidate(ObMemberList& mlist) const int ObElection::get_valid_candidate(ObMemberList &mlist) const
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -1503,8 +1503,8 @@ int ObElection::get_valid_candidate(ObMemberList& mlist) const
return ret; return ret;
} }
int ObElection::get_leader(ObAddr& leader, int64_t& leader_epoch, bool& is_elected_by_changing_leader, int ObElection::get_leader(ObAddr &leader, int64_t &leader_epoch, bool &is_elected_by_changing_leader,
ObTsWindows& changing_leader_windows) const ObTsWindows &changing_leader_windows) const
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -1534,7 +1534,7 @@ int ObElection::get_leader(ObAddr& leader, int64_t& leader_epoch, bool& is_elect
} }
int ObElection::get_leader( int ObElection::get_leader(
ObAddr& leader, ObAddr& previous_leader, int64_t& leader_epoch, bool& is_elected_by_changing_leader) const ObAddr &leader, ObAddr &previous_leader, int64_t &leader_epoch, bool &is_elected_by_changing_leader) const
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -1561,7 +1561,7 @@ int ObElection::get_leader(
return ret; return ret;
} }
int ObElection::get_current_leader(ObAddr& leader) const int ObElection::get_current_leader(ObAddr &leader) const
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -1578,7 +1578,7 @@ int ObElection::get_current_leader(ObAddr& leader) const
return ret; return ret;
} }
int ObElection::get_election_info(ObElectionInfo& election_info) const int ObElection::get_election_info(ObElectionInfo &election_info) const
{ {
RLockGuard guard(lock_); RLockGuard guard(lock_);
election_info = *this; election_info = *this;
@ -1651,7 +1651,7 @@ int ObElection::cal_valid_candidates_()
} }
// check if the corresponding lease is valid // check if the corresponding lease is valid
bool ObElection::verify_lease_start_valid_(const ObAddr& leader, const int64_t logic_lease_start_t1) const bool ObElection::verify_lease_start_valid_(const ObAddr &leader, const int64_t logic_lease_start_t1) const
{ {
bool bool_ret = false; bool bool_ret = false;
if (!leader.is_valid() || logic_lease_start_t1 <= 0) { if (!leader.is_valid() || logic_lease_start_t1 <= 0) {
@ -1671,7 +1671,7 @@ bool ObElection::verify_lease_start_valid_(const ObAddr& leader, const int64_t l
// call this interface to assign a leader // call this interface to assign a leader
// called when partition is created // called when partition is created
int ObElection::leader_takeover(const ObAddr& leader, const int64_t lease_start, int64_t& leader_epoch) int ObElection::leader_takeover(const ObAddr &leader, const int64_t lease_start, int64_t &leader_epoch)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
const int64_t logic_t1 = ((lease_start - get_election_time_offset()) / T_ELECT2) * T_ELECT2; const int64_t logic_t1 = ((lease_start - get_election_time_offset()) / T_ELECT2) * T_ELECT2;
@ -1730,7 +1730,7 @@ int ObElection::force_leader_async_()
return ret; return ret;
} }
int ObElection::change_leader_async_(const ObAddr& server, ObTsWindows& changing_leader_windows) int ObElection::change_leader_async_(const ObAddr &server, ObTsWindows &changing_leader_windows)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
int64_t delay = 0; int64_t delay = 0;
@ -1810,7 +1810,7 @@ int ObElection::update_info_from_eg_()
const int64_t cur_ts = get_current_ts(); const int64_t cur_ts = get_current_ts();
if (OB_FAIL(election_group_->get_leader_lease_info( if (OB_FAIL(election_group_->get_leader_lease_info(
partition_, eg_part_array_idx_, cur_ts, eg_lease_end, eg_takeover_t1_ts))) { partition_, eg_part_array_idx_, cur_ts, eg_lease_end, eg_takeover_t1_ts))) {
FORCE_ELECT_LOG(WARN, "get_leader_lease_info failed", K(ret), K_(partition), "eleciton", *this); FORCE_ELECT_LOG(WARN, "get_leader_lease_info failed", K(ret), K_(partition), "election", *this);
} else if (eg_lease_end < 0) { } else if (eg_lease_end < 0) {
ret = OB_ERR_UNEXPECTED; ret = OB_ERR_UNEXPECTED;
FORCE_ELECT_LOG(ERROR, "eg_lease_end is invalid", K(ret), K(eg_lease_end), "election", *this); FORCE_ELECT_LOG(ERROR, "eg_lease_end is invalid", K(ret), K(eg_lease_end), "election", *this);
@ -1847,7 +1847,7 @@ void ObElection::change_leader_clear_()
} }
// check if server is leader, check role and lease also // check if server is leader, check role and lease also
bool ObElection::is_real_leader_(const ObAddr& server) const bool ObElection::is_real_leader_(const ObAddr &server) const
{ {
bool bool_ret = false; bool bool_ret = false;
const int64_t cur_ts = get_current_ts(); const int64_t cur_ts = get_current_ts();
@ -1879,7 +1879,7 @@ bool ObElection::check_if_allowed_to_move_into_eg_()
return bool_ret; return bool_ret;
} }
int ObElection::leader_elected_(const ObAddr& leader, const int64_t logic_T1_timestamp, const int64_t lease_time, int ObElection::leader_elected_(const ObAddr &leader, const int64_t logic_T1_timestamp, const int64_t lease_time,
const bool is_elected_by_changing_leader) const bool is_elected_by_changing_leader)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -1959,7 +1959,7 @@ int ObElection::leader_elected_(const ObAddr& leader, const int64_t logic_T1_tim
return ret; return ret;
} }
int ObElection::leader_takeover_(const ObAddr& leader, const int64_t epoch, const int64_t logic_t1, int ObElection::leader_takeover_(const ObAddr &leader, const int64_t epoch, const int64_t logic_t1,
const int64_t lease_time, const bool is_elected_by_changing_leader) const int64_t lease_time, const bool is_elected_by_changing_leader)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -1995,7 +1995,7 @@ int ObElection::leader_takeover_(const ObAddr& leader, const int64_t epoch, cons
return ret; return ret;
} }
int ObElection::leader_reappoint_(const ObAddr& leader, const int64_t logic_t1, const int64_t lease_time) int ObElection::leader_reappoint_(const ObAddr &leader, const int64_t logic_t1, const int64_t lease_time)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
const int64_t cur_ts = get_current_ts(); const int64_t cur_ts = get_current_ts();
@ -2032,7 +2032,7 @@ int ObElection::verify_leader_valid_(const int64_t cur_ts) const
return ret; return ret;
} }
int ObElection::get_leader_(ObAddr& leader) const int ObElection::get_leader_(ObAddr &leader) const
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
const int64_t cur_ts = get_current_ts(); const int64_t cur_ts = get_current_ts();
@ -2185,12 +2185,12 @@ bool ObElection::lease_expired_too_long_(const int64_t cur_ts) const
return (cur_ts - leader_revoke_timestamp_ > EXPIRED_TIME_FOR_WARNING); return (cur_ts - leader_revoke_timestamp_ > EXPIRED_TIME_FOR_WARNING);
} }
void ObElection::set_unconfirmed_leader_(const ObAddr& unconfirmed_leader) void ObElection::set_unconfirmed_leader_(const ObAddr &unconfirmed_leader)
{ {
unconfirmed_leader_ = unconfirmed_leader; unconfirmed_leader_ = unconfirmed_leader;
} }
int ObElection::move_out_election_group(const ObElectionGroupId& eg_id) int ObElection::move_out_election_group(const ObElectionGroupId &eg_id)
{ {
// called from election group // called from election group
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -2261,16 +2261,16 @@ int ObElection::move_out_election_group_unlock_()
return ret; return ret;
} }
int ObElection::move_into_election_group(const ObElectionGroupId& eg_id) int ObElection::move_into_election_group(const ObElectionGroupId &eg_id)
{ {
WLockGuard guard(lock_); WLockGuard guard(lock_);
return move_into_election_group_unlock_(eg_id); return move_into_election_group_unlock_(eg_id);
} }
int ObElection::move_into_election_group_unlock_(const ObElectionGroupId& eg_id) int ObElection::move_into_election_group_unlock_(const ObElectionGroupId &eg_id)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
ObIElectionGroup* election_group = NULL; ObIElectionGroup *election_group = NULL;
if (!is_inited_) { if (!is_inited_) {
ret = OB_NOT_INIT; ret = OB_NOT_INIT;
@ -2292,7 +2292,7 @@ int ObElection::move_into_election_group_unlock_(const ObElectionGroupId& eg_id)
} }
} else if (!current_leader_.is_valid() || current_leader_ != eg_id.get_server() || } else if (!current_leader_.is_valid() || current_leader_ != eg_id.get_server() ||
!is_real_leader_(current_leader_)) { !is_real_leader_(current_leader_)) {
// allow follower moive in only if leader is valid and match group's leader // allow follower move in only if leader is valid and match group's leader
ret = OB_STATE_NOT_MATCH; ret = OB_STATE_NOT_MATCH;
if (REACH_TIME_INTERVAL(100 * 1000)) { if (REACH_TIME_INTERVAL(100 * 1000)) {
FORCE_ELECT_LOG(WARN, "current_leader_ not match with eg_id", K(ret), K(eg_id), "election", *this); FORCE_ELECT_LOG(WARN, "current_leader_ not match with eg_id", K(ret), K(eg_id), "election", *this);
@ -2502,7 +2502,7 @@ bool ObElection::can_elect_new_leader_() const
bool_ret = false; bool_ret = false;
} else if (common::PRIMARY_CLUSTER == cluster_type || } else if (common::PRIMARY_CLUSTER == cluster_type ||
ObMultiClusterUtil::is_cluster_private_table(partition_.get_table_id())) { ObMultiClusterUtil::is_cluster_private_table(partition_.get_table_id())) {
// main datebase or private table allow elect // main database or private table allow elect
bool_ret = true; bool_ret = true;
} else if (share::OBSERVER_INVALID_STATUS == server_status) { } else if (share::OBSERVER_INVALID_STATUS == server_status) {
// server status unknown, non-private table not allow to elect // server status unknown, non-private table not allow to elect
@ -2538,7 +2538,7 @@ void ObElection::run_gt1_task(const int64_t real_T1_timestamp)
election_time_offset_ = new_election_time_offset_; election_time_offset_ = new_election_time_offset_;
temp_election_time_offset_ = election_time_offset_; temp_election_time_offset_ = election_time_offset_;
takeover_t1_timestamp_ = 0; takeover_t1_timestamp_ = 0;
ELECT_ASYNC_LOG(INFO, "be dead, parameters conversion is compeleted", "election", *this); ELECT_ASYNC_LOG(INFO, "be dead, parameters conversion is completed", "election", *this);
} else { } else {
} }
/*********************************************************************************************/ /*********************************************************************************************/
@ -2902,7 +2902,7 @@ void ObElection::run_gt1_task(const int64_t real_T1_timestamp)
/*********************************************************************************************/ /*********************************************************************************************/
} }
int ObElection::send_devote_prepare_(const ObElectionMsgDEPrepare& msg) int ObElection::send_devote_prepare_(const ObElectionMsgDEPrepare &msg)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -2927,7 +2927,7 @@ int ObElection::send_devote_prepare_(const ObElectionMsgDEPrepare& msg)
} }
// add lease time, make sure follower won't vote for others // add lease time, make sure follower won't vote for others
int ObElection::send_vote_prepare_(const ObElectionMsgPrepare& msg) int ObElection::send_vote_prepare_(const ObElectionMsgPrepare &msg)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
ObAddr new_leader; ObAddr new_leader;
@ -2951,7 +2951,7 @@ int ObElection::send_vote_prepare_(const ObElectionMsgPrepare& msg)
return ret; return ret;
} }
int ObElection::send_change_leader_vote_prepare_(const ObElectionMsgPrepare& msg) int ObElection::send_change_leader_vote_prepare_(const ObElectionMsgPrepare &msg)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -3148,7 +3148,7 @@ void ObElection::run_gt2_task(const int64_t real_T2_timestamp)
} }
} }
int ObElection::get_decentralized_candidate_(ObAddr& server, ObElectionPriority& priority, int64_t& lease_time) int ObElection::get_decentralized_candidate_(ObAddr &server, ObElectionPriority &priority, int64_t &lease_time)
{ {
int ret = msg_pool_.get_decentralized_candidate(server, priority, replica_num_, T1_timestamp_, lease_time); int ret = msg_pool_.get_decentralized_candidate(server, priority, replica_num_, T1_timestamp_, lease_time);
if (OB_FAIL(ret)) { if (OB_FAIL(ret)) {
@ -3157,7 +3157,7 @@ int ObElection::get_decentralized_candidate_(ObAddr& server, ObElectionPriority&
return ret; return ret;
} }
int ObElection::get_centralized_candidate_(ObAddr& cur_leader, ObAddr& new_leader, int64_t msg_t1) int ObElection::get_centralized_candidate_(ObAddr &cur_leader, ObAddr &new_leader, int64_t msg_t1)
{ {
int ret = msg_pool_.get_centralized_candidate(cur_leader, new_leader, msg_t1); int ret = msg_pool_.get_centralized_candidate(cur_leader, new_leader, msg_t1);
if (OB_FAIL(ret)) { if (OB_FAIL(ret)) {
@ -3166,7 +3166,7 @@ int ObElection::get_centralized_candidate_(ObAddr& cur_leader, ObAddr& new_leade
return ret; return ret;
} }
int ObElection::send_devote_vote_(const ObAddr& server, const ObElectionMsgDEVote& msg) int ObElection::send_devote_vote_(const ObAddr &server, const ObElectionMsgDEVote &msg)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -3281,7 +3281,7 @@ int ObElection::check_centralized_majority_()
} else if (is_candidate_(self_) && is_real_leader_(self_) && cur_leader == current_leader_) { } else if (is_candidate_(self_) && is_real_leader_(self_) && cur_leader == current_leader_) {
bool change_leader = false; bool change_leader = false;
if (cur_leader != new_leader) { // change leader if (cur_leader != new_leader) { // change leader
// need new leader's message and its prioeity // need new leader's message and its priority
if (!new_leader_priority.is_valid()) { if (!new_leader_priority.is_valid()) {
ret = OB_ERR_UNEXPECTED; ret = OB_ERR_UNEXPECTED;
FORCE_ELECT_LOG(WARN, "change leader but new leader priority is invalid", K(ret), K(new_leader_priority)); FORCE_ELECT_LOG(WARN, "change leader but new leader priority is invalid", K(ret), K(new_leader_priority));
@ -3329,7 +3329,7 @@ int ObElection::check_centralized_majority_()
return ret; return ret;
} }
int ObElection::send_vote_vote_(const ObAddr& server, const ObElectionMsgVote& msg) int ObElection::send_vote_vote_(const ObAddr &server, const ObElectionMsgVote &msg)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -3348,7 +3348,7 @@ int ObElection::send_vote_vote_(const ObAddr& server, const ObElectionMsgVote& m
return ret; return ret;
} }
int ObElection::notify_leader_(const ObAddr& server) int ObElection::notify_leader_(const ObAddr &server)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
int64_t lease_time = 0; int64_t lease_time = 0;
@ -3448,7 +3448,7 @@ void ObElection::run_gt3_task(const int64_t real_T3_timestamp)
} }
} }
int ObElection::check_decentralized_majority_(ObAddr& new_leader, int64_t& ticket) int ObElection::check_decentralized_majority_(ObAddr &new_leader, int64_t &ticket)
{ {
int ret = msg_pool_.check_decentralized_majority(new_leader, ticket, replica_num_, T1_timestamp_); int ret = msg_pool_.check_decentralized_majority(new_leader, ticket, replica_num_, T1_timestamp_);
if (OB_FAIL(ret)) { if (OB_FAIL(ret)) {
@ -3458,7 +3458,7 @@ int ObElection::check_decentralized_majority_(ObAddr& new_leader, int64_t& ticke
} }
// return OB_SUCCESS, avoid no leader // return OB_SUCCESS, avoid no leader
int ObElection::collect_valid_candidates_(const ObElectionMsgVote& msg) int ObElection::collect_valid_candidates_(const ObElectionMsgVote &msg)
{ {
int tmp_ret = OB_SUCCESS; int tmp_ret = OB_SUCCESS;
@ -3487,7 +3487,7 @@ int ObElection::collect_valid_candidates_(const ObElectionMsgVote& msg)
return OB_SUCCESS; return OB_SUCCESS;
} }
int ObElection::send_devote_success_(const ObElectionMsgDESuccess& msg) int ObElection::send_devote_success_(const ObElectionMsgDESuccess &msg)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -3506,7 +3506,7 @@ int ObElection::send_devote_success_(const ObElectionMsgDESuccess& msg)
return ret; return ret;
} }
int ObElection::send_vote_success_(const ObElectionMsgSuccess& msg) int ObElection::send_vote_success_(const ObElectionMsgSuccess &msg)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -3620,7 +3620,7 @@ void ObElection::run_change_leader_task(const int64_t real_T1_timestamp)
} }
int ObElection::post_election_msg_( int ObElection::post_election_msg_(
const common::ObMemberList& mlist, const common::ObPartitionKey& partition, const ObElectionMsg& msg) const common::ObMemberList &mlist, const common::ObPartitionKey &partition, const ObElectionMsg &msg)
{ {
int ret = common::OB_SUCCESS; int ret = common::OB_SUCCESS;
common::ObAddr server; common::ObAddr server;
@ -3653,7 +3653,7 @@ int ObElection::post_election_msg_(
} }
int ObElection::post_election_msg_( int ObElection::post_election_msg_(
const common::ObAddr& server, const common::ObPartitionKey& partition, const ObElectionMsg& msg) const common::ObAddr &server, const common::ObPartitionKey &partition, const ObElectionMsg &msg)
{ {
int ret = common::OB_SUCCESS; int ret = common::OB_SUCCESS;
const int64_t dst_cluster_id = obrpc::ObRpcNetHandler::CLUSTER_ID; const int64_t dst_cluster_id = obrpc::ObRpcNetHandler::CLUSTER_ID;
@ -3694,7 +3694,7 @@ bool ObElection::in_reappoint_period_(const int64_t ts)
return bool_ret; return bool_ret;
} }
void ObElection::get_change_leader_window_(const int64_t ts, ObTsWindows& windows) const void ObElection::get_change_leader_window_(const int64_t ts, ObTsWindows &windows) const
{ {
const int64_t T_SWITCH_LEADER_RESERVED = ObServerConfig::get_instance().trx_force_kill_threshold; const int64_t T_SWITCH_LEADER_RESERVED = ObServerConfig::get_instance().trx_force_kill_threshold;
if (!GCONF.enable_smooth_leader_switch) { if (!GCONF.enable_smooth_leader_switch) {
@ -3704,7 +3704,7 @@ void ObElection::get_change_leader_window_(const int64_t ts, ObTsWindows& window
} }
} }
int ObElection::get_priority(ObElectionPriority& priority) const int ObElection::get_priority(ObElectionPriority &priority) const
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -3719,7 +3719,7 @@ int ObElection::get_priority(ObElectionPriority& priority) const
return ret; return ret;
} }
int ObElection::get_timestamp(int64_t& gts, ObAddr& leader) const int ObElection::get_timestamp(int64_t &gts, ObAddr &leader) const
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
RLockGuard guard(lock_); RLockGuard guard(lock_);
@ -3755,20 +3755,20 @@ int ObElection::get_timestamp(int64_t& gts, ObAddr& leader) const
return ret; return ret;
} }
int64_t ObElection::get_msg_real_T1_(const ObElectionVoteMsg& msg) const int64_t ObElection::get_msg_real_T1_(const ObElectionVoteMsg &msg) const
{ {
int64_t result_ts = OB_INVALID_TIMESTAMP; int64_t result_ts = OB_INVALID_TIMESTAMP;
// is msg is about change leader, T1 is strictly integer multiples of T_ELECT2 // is msg is about change leader, T1 is strictly integer multiples of T_ELECT2
// otherwise, just reduce election_time_offset_ // otherwise, just reduce election_time_offset_
if ((msg.get_msg_type() == ObElectionMsgType::OB_ELECTION_VOTE_PREPARE && if ((msg.get_msg_type() == ObElectionMsgType::OB_ELECTION_VOTE_PREPARE &&
static_cast<const ObElectionMsgPrepare&>(msg).get_cur_leader() != static_cast<const ObElectionMsgPrepare &>(msg).get_cur_leader() !=
static_cast<const ObElectionMsgPrepare&>(msg).get_new_leader()) || static_cast<const ObElectionMsgPrepare &>(msg).get_new_leader()) ||
(msg.get_msg_type() == ObElectionMsgType::OB_ELECTION_VOTE_VOTE && (msg.get_msg_type() == ObElectionMsgType::OB_ELECTION_VOTE_VOTE &&
static_cast<const ObElectionMsgVote&>(msg).get_cur_leader() != static_cast<const ObElectionMsgVote &>(msg).get_cur_leader() !=
static_cast<const ObElectionMsgVote&>(msg).get_new_leader()) || static_cast<const ObElectionMsgVote &>(msg).get_new_leader()) ||
(msg.get_msg_type() == ObElectionMsgType::OB_ELECTION_VOTE_SUCCESS && (msg.get_msg_type() == ObElectionMsgType::OB_ELECTION_VOTE_SUCCESS &&
static_cast<const ObElectionMsgSuccess&>(msg).get_cur_leader() != static_cast<const ObElectionMsgSuccess &>(msg).get_cur_leader() !=
static_cast<const ObElectionMsgSuccess&>(msg).get_new_leader())) { static_cast<const ObElectionMsgSuccess &>(msg).get_new_leader())) {
result_ts = (msg.get_T1_timestamp() + election_time_offset_) / T_ELECT2 * T_ELECT2; result_ts = (msg.get_T1_timestamp() + election_time_offset_) / T_ELECT2 * T_ELECT2;
} else { } else {
result_ts = real_ts_(msg.get_T1_timestamp()); result_ts = real_ts_(msg.get_T1_timestamp());
@ -3787,7 +3787,7 @@ int64_t ObElection::logic_ts_(int64_t real_ts) const
} }
/**********should removed after a barrier version bigger than 3.1**********/ /**********should removed after a barrier version bigger than 3.1**********/
void ObElection::insert_physical_condition_into_msg_(const ObElectionMsg& msg) void ObElection::insert_physical_condition_into_msg_(const ObElectionMsg &msg)
{ {
if (OB_SYS_TENANT_ID != extract_tenant_id(partition_.get_table_id())) { if (OB_SYS_TENANT_ID != extract_tenant_id(partition_.get_table_id())) {
int64_t send_timestamp = msg.get_send_timestamp(); int64_t send_timestamp = msg.get_send_timestamp();
@ -3811,12 +3811,12 @@ void ObElection::insert_physical_condition_into_msg_(const ObElectionMsg& msg)
cond <<= 61; cond <<= 61;
cond &= 0x6000000000000000; cond &= 0x6000000000000000;
send_timestamp |= cond; send_timestamp |= cond;
const_cast<ObElectionMsg&>(msg).set_send_timestamp(send_timestamp); const_cast<ObElectionMsg &>(msg).set_send_timestamp(send_timestamp);
} else { } else {
} }
} }
PhysicalCondition ObElection::fetch_others_condition_(const ObElectionMsg& msg) PhysicalCondition ObElection::fetch_others_condition_(const ObElectionMsg &msg)
{ {
int64_t send_timestamp = msg.get_send_timestamp(); int64_t send_timestamp = msg.get_send_timestamp();
int64_t condition = send_timestamp & 0x6000000000000000; int64_t condition = send_timestamp & 0x6000000000000000;
@ -3841,7 +3841,7 @@ PhysicalCondition ObElection::fetch_others_condition_(const ObElectionMsg& msg)
break; break;
} }
send_timestamp &= 0x9fffffffffffffff; send_timestamp &= 0x9fffffffffffffff;
const_cast<ObElectionMsg&>(msg).set_send_timestamp(send_timestamp); const_cast<ObElectionMsg &>(msg).set_send_timestamp(send_timestamp);
return ret; return ret;
} }
@ -3859,7 +3859,7 @@ void ObElection::register_gt1_with_neccessary_retry_(int64_t next_t1_timestamp)
} while (OB_INVALID_ARGUMENT == temp_ret && delay < 0); } while (OB_INVALID_ARGUMENT == temp_ret && delay < 0);
} }
bool ObElection::update_condition_(const ObElectionMsg& msg, const int64_t msg_t1) bool ObElection::update_condition_(const ObElectionMsg &msg, const int64_t msg_t1)
{ {
PhysicalCondition cond = fetch_others_condition_(msg); PhysicalCondition cond = fetch_others_condition_(msg);
if (physical_condition_ < cond) { if (physical_condition_ < cond) {

140
src/election/ob_election_group.cpp
View File

@ -38,10 +38,10 @@ using namespace common;
namespace election { namespace election {
int ObElectionGroup::init(const ObElectionGroupId& eg_id, const ObAddr& self, const uint64_t tenant_id, int ObElectionGroup::init(const ObElectionGroupId &eg_id, const ObAddr &self, const uint64_t tenant_id,
const common::ObMemberList& sorted_member_list, const int64_t replica_num, ObIElectionRpc* rpc, ObTimeWheel* tw, const common::ObMemberList &sorted_member_list, const int64_t replica_num, ObIElectionRpc *rpc, ObTimeWheel *tw,
ObIElectionMgr* election_mgr, ObElectionGroupCache* election_group_cache, ObIElectionGroupPriorityGetter* eg_cb, ObIElectionMgr *election_mgr, ObElectionGroupCache *election_group_cache, ObIElectionGroupPriorityGetter *eg_cb,
ObIElectionGroupMgr* eg_mgr) ObIElectionGroupMgr *eg_mgr)
{ {
// init for leader election group // init for leader election group
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -97,11 +97,11 @@ int ObElectionGroup::init(const ObElectionGroupId& eg_id, const ObAddr& self, co
return ret; return ret;
} }
int ObElectionGroup::init(const ObElectionGroupId& eg_id, const ObAddr& self, const uint64_t tenant_id, int ObElectionGroup::init(const ObElectionGroupId &eg_id, const ObAddr &self, const uint64_t tenant_id,
ObIElectionRpc* rpc, ObTimeWheel* tw, ObIElectionMgr* election_mgr, ObElectionGroupCache* election_group_cache, ObIElectionRpc *rpc, ObTimeWheel *tw, ObIElectionMgr *election_mgr, ObElectionGroupCache *election_group_cache,
ObIElectionGroupPriorityGetter* eg_cb, ObIElectionGroupMgr* eg_mgr) ObIElectionGroupPriorityGetter *eg_cb, ObIElectionGroupMgr *eg_mgr)
{ {
// init for follower elction group // init for follower election group
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
if (is_inited_) { if (is_inited_) {
@ -331,7 +331,7 @@ int ObElectionGroup::try_stop_()
return ret; return ret;
} }
int ObElectionGroup::try_update_eg_version(const int64_t msg_eg_version, const ObPartitionArray& partition_array) int ObElectionGroup::try_update_eg_version(const int64_t msg_eg_version, const ObPartitionArray &partition_array)
{ {
// update eg_version // update eg_version
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -357,10 +357,10 @@ bool ObElectionGroup::is_need_check_eg_version(const int64_t msg_eg_version) con
return (ROLE_LEADER != ATOMIC_LOAD(&role_) && (ATOMIC_LOAD(&eg_version_) != msg_eg_version)); return (ROLE_LEADER != ATOMIC_LOAD(&role_) && (ATOMIC_LOAD(&eg_version_) != msg_eg_version));
} }
int ObElectionGroup::check_eg_version(const int64_t msg_eg_version, const ObPartitionArray& partition_array, int ObElectionGroup::check_eg_version(const int64_t msg_eg_version, const ObPartitionArray &partition_array,
ObPartitionArray& pending_move_out_array, ObPartitionArray& pending_move_in_array) ObPartitionArray &pending_move_out_array, ObPartitionArray &pending_move_in_array)
{ {
// check version before follower group process messsages // check version before follower group process messages
// is_running may be false for follower group // is_running may be false for follower group
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -420,7 +420,7 @@ int ObElectionGroup::check_eg_version(const int64_t msg_eg_version, const ObPart
return ret; return ret;
} }
bool ObElectionGroup::is_part_array_equal_(const ObPartitionArray& partition_array) const bool ObElectionGroup::is_part_array_equal_(const ObPartitionArray &partition_array) const
{ {
bool bool_ret = false; bool bool_ret = false;
if (partition_array_.count() != partition_array.count()) { if (partition_array_.count() != partition_array.count()) {
@ -439,8 +439,8 @@ bool ObElectionGroup::is_part_array_equal_(const ObPartitionArray& partition_arr
return bool_ret; return bool_ret;
} }
int ObElectionGroup::gen_pending_move_part_array_(const int64_t msg_eg_version, const ObPartitionArray& partition_array, int ObElectionGroup::gen_pending_move_part_array_(const int64_t msg_eg_version, const ObPartitionArray &partition_array,
ObPartitionArray& pending_move_in_array, ObPartitionArray& pending_move_out_array) ObPartitionArray &pending_move_in_array, ObPartitionArray &pending_move_out_array)
{ {
// follower group calculate move in/out list according to leader group's message // follower group calculate move in/out list according to leader group's message
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -456,7 +456,7 @@ int ObElectionGroup::gen_pending_move_part_array_(const int64_t msg_eg_version,
const int64_t self_part_count = partition_array_.count(); const int64_t self_part_count = partition_array_.count();
int64_t tmp_idx = -1; int64_t tmp_idx = -1;
for (int64_t i = 0; i < self_part_count; ++i) { for (int64_t i = 0; i < self_part_count; ++i) {
const ObPartitionKey& tmp_arg_pkey = partition_array_.at(i); const ObPartitionKey &tmp_arg_pkey = partition_array_.at(i);
if (!is_pkey_in_array_(tmp_arg_pkey, partition_array, tmp_idx)) { if (!is_pkey_in_array_(tmp_arg_pkey, partition_array, tmp_idx)) {
if (OB_FAIL(pending_move_out_array.push_back(tmp_arg_pkey))) { if (OB_FAIL(pending_move_out_array.push_back(tmp_arg_pkey))) {
FORCE_ELECT_LOG(WARN, "array push_back failed", K(ret), K(tmp_arg_pkey)); FORCE_ELECT_LOG(WARN, "array push_back failed", K(ret), K(tmp_arg_pkey));
@ -466,7 +466,7 @@ int ObElectionGroup::gen_pending_move_part_array_(const int64_t msg_eg_version,
// calculate move in partitions // calculate move in partitions
const int64_t arg_part_count = partition_array.count(); const int64_t arg_part_count = partition_array.count();
for (int64_t i = 0; i < arg_part_count; ++i) { for (int64_t i = 0; i < arg_part_count; ++i) {
const ObPartitionKey& tmp_arg_pkey = partition_array.at(i); const ObPartitionKey &tmp_arg_pkey = partition_array.at(i);
if (!is_partition_exist_unlock_(tmp_arg_pkey, tmp_idx)) { if (!is_partition_exist_unlock_(tmp_arg_pkey, tmp_idx)) {
if (OB_FAIL(pending_move_in_array.push_back(tmp_arg_pkey))) { if (OB_FAIL(pending_move_in_array.push_back(tmp_arg_pkey))) {
FORCE_ELECT_LOG(WARN, "array push_back failed", K(ret), K(tmp_arg_pkey)); FORCE_ELECT_LOG(WARN, "array push_back failed", K(ret), K(tmp_arg_pkey));
@ -477,7 +477,7 @@ int ObElectionGroup::gen_pending_move_part_array_(const int64_t msg_eg_version,
return ret; return ret;
} }
int ObElectionGroup::batch_move_out_partition_(const ObPartitionArray& pending_move_out_array) int ObElectionGroup::batch_move_out_partition_(const ObPartitionArray &pending_move_out_array)
{ {
// handle with move out partitions, // handle with move out partitions,
// no lock protected, call interface of ObElection // no lock protected, call interface of ObElection
@ -524,7 +524,7 @@ int ObElectionGroup::push_lease_to_part_stat_()
return push_lease_to_part_skip_pkey_(tmp_pkey); return push_lease_to_part_skip_pkey_(tmp_pkey);
} }
int ObElectionGroup::push_lease_to_part_skip_pkey_(const ObPartitionKey& skip_pkey) int ObElectionGroup::push_lease_to_part_skip_pkey_(const ObPartitionKey &skip_pkey)
{ {
// push group's lease to every partition's state, except skip_pkey // push group's lease to every partition's state, except skip_pkey
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -550,7 +550,7 @@ int ObElectionGroup::push_lease_to_part_skip_pkey_(const ObPartitionKey& skip_pk
} }
int ObElectionGroup::move_in_partition( int ObElectionGroup::move_in_partition(
const ObPartitionKey& pkey, const lease_t part_lease, const int64_t takeover_t1_timestamp, int64_t& ret_idx) const ObPartitionKey &pkey, const lease_t part_lease, const int64_t takeover_t1_timestamp, int64_t &ret_idx)
{ {
// move in partition, called by ObElection // move in partition, called by ObElection
// do:add partitions;update group's lease_end. // do:add partitions;update group's lease_end.
@ -618,7 +618,7 @@ int ObElectionGroup::move_in_partition(
return ret; return ret;
} }
int ObElectionGroup::move_out_partition(const ObPartitionKey& pkey, const int64_t target_idx) int ObElectionGroup::move_out_partition(const ObPartitionKey &pkey, const int64_t target_idx)
{ {
// move out partition, called by ObElection // move out partition, called by ObElection
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -724,7 +724,7 @@ bool ObElectionGroup::is_empty_() const
return (partition_array_.count() == 0); return (partition_array_.count() == 0);
} }
bool ObElectionGroup::is_pkey_exist_(const ObPartitionKey& pkey) const bool ObElectionGroup::is_pkey_exist_(const ObPartitionKey &pkey) const
{ {
// read lock added here // read lock added here
int64_t dummy_idx = -1; int64_t dummy_idx = -1;
@ -732,13 +732,13 @@ bool ObElectionGroup::is_pkey_exist_(const ObPartitionKey& pkey) const
return is_partition_exist_unlock_(pkey, dummy_idx); return is_partition_exist_unlock_(pkey, dummy_idx);
} }
bool ObElectionGroup::is_partition_exist_unlock_(const ObPartitionKey& pkey, int64_t& idx) const bool ObElectionGroup::is_partition_exist_unlock_(const ObPartitionKey &pkey, int64_t &idx) const
{ {
return is_pkey_in_array_(pkey, partition_array_, idx); return is_pkey_in_array_(pkey, partition_array_, idx);
} }
bool ObElectionGroup::is_pkey_in_array_( bool ObElectionGroup::is_pkey_in_array_(
const ObPartitionKey& pkey, const ObPartitionArray& partition_array, int64_t& idx) const const ObPartitionKey &pkey, const ObPartitionArray &partition_array, int64_t &idx) const
{ {
bool bool_ret = false; bool bool_ret = false;
idx = -1; idx = -1;
@ -757,7 +757,7 @@ int64_t ObElectionGroup::get_current_ts() const
return ObTimeUtility::current_time(); return ObTimeUtility::current_time();
} }
int ObElectionGroup::process_vote_prepare_(const ObElectionMsgEGPrepare& msg, const int64_t msg_eg_version) int ObElectionGroup::process_vote_prepare_(const ObElectionMsgEGPrepare &msg, const int64_t msg_eg_version)
{ {
// caller need check and add read lock // caller need check and add read lock
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -825,7 +825,7 @@ int ObElectionGroup::process_vote_prepare_(const ObElectionMsgEGPrepare& msg, co
return ret; return ret;
} }
int ObElectionGroup::local_handle_vote_prepare_(const ObElectionMsgEGPrepare& msg) int ObElectionGroup::local_handle_vote_prepare_(const ObElectionMsgEGPrepare &msg)
{ {
// caller need add read lock // caller need add read lock
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -859,7 +859,7 @@ int ObElectionGroup::local_handle_vote_prepare_(const ObElectionMsgEGPrepare& ms
} }
int ObElectionGroup::handle_vote_prepare( int ObElectionGroup::handle_vote_prepare(
const ObElectionMsgEGPrepare& msg, const int64_t msg_eg_version, obrpc::ObElectionRpcResult& result) const ObElectionMsgEGPrepare &msg, const int64_t msg_eg_version, obrpc::ObElectionRpcResult &result)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -944,7 +944,7 @@ int ObElectionGroup::try_centralized_voting_(const int64_t lease_time)
return ret; return ret;
} }
int ObElectionGroup::local_handle_vote_vote_(const ObElectionMsgEGVote& msg) int ObElectionGroup::local_handle_vote_vote_(const ObElectionMsgEGVote &msg)
{ {
// caller need add read lock // caller need add read lock
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -1006,9 +1006,9 @@ int ObElectionGroup::local_handle_vote_vote_(const ObElectionMsgEGVote& msg)
return ret; return ret;
} }
int ObElectionGroup::handle_vote_vote(const ObElectionMsgEGVote& msg, const int64_t msg_eg_version, int ObElectionGroup::handle_vote_vote(const ObElectionMsgEGVote &msg, const int64_t msg_eg_version,
const bool is_array_deserialized, const char* buf, const int64_t buf_len, const int64_t array_start_pos, const bool is_array_deserialized, const char *buf, const int64_t buf_len, const int64_t array_start_pos,
ObPartitionArray& msg_part_array, obrpc::ObElectionRpcResult& result) ObPartitionArray &msg_part_array, obrpc::ObElectionRpcResult &result)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -1107,7 +1107,7 @@ int ObElectionGroup::handle_vote_vote(const ObElectionMsgEGVote& msg, const int6
return ret; return ret;
} }
int ObElectionGroup::send_vote_vote_(const ObElectionMsgEGVote& msg) int ObElectionGroup::send_vote_vote_(const ObElectionMsgEGVote &msg)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
const int64_t cur_ts = get_current_ts(); const int64_t cur_ts = get_current_ts();
@ -1125,7 +1125,7 @@ int ObElectionGroup::send_vote_vote_(const ObElectionMsgEGVote& msg)
return ret; return ret;
} }
int ObElectionGroup::get_majority_part_idx_array_(ObPartIdxArray& majority_part_idx_array) int ObElectionGroup::get_majority_part_idx_array_(ObPartIdxArray &majority_part_idx_array)
{ {
// get the list of partitions who received majority votes // get the list of partitions who received majority votes
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -1174,7 +1174,7 @@ int ObElectionGroup::check_centralized_majority_()
new_leader, new_leader,
is_all_part_merged_in_, is_all_part_merged_in_,
is_eg_majority, is_eg_majority,
part_state_array_, // for counting every singal partition's tick part_state_array_, // for counting every single partition's tick
partition_array_, partition_array_,
eg_version_, eg_version_,
replica_num_, replica_num_,
@ -1237,8 +1237,8 @@ int ObElectionGroup::check_centralized_majority_()
return ret; return ret;
} }
bool ObElectionGroup::is_pkey_in_majority_array_(const ObPartitionKey& tmp_pkey, const ObPartitionArray& msg_part_array, bool ObElectionGroup::is_pkey_in_majority_array_(const ObPartitionKey &tmp_pkey, const ObPartitionArray &msg_part_array,
const ObPartIdxArray& msg_majority_part_idx_array) const const ObPartIdxArray &msg_majority_part_idx_array) const
{ {
bool bool_ret = false; bool bool_ret = false;
const int64_t msg_majority_array_cnt = msg_majority_part_idx_array.count(); const int64_t msg_majority_array_cnt = msg_majority_part_idx_array.count();
@ -1257,8 +1257,8 @@ bool ObElectionGroup::is_pkey_in_majority_array_(const ObPartitionKey& tmp_pkey,
} }
int ObElectionGroup::parse_majority_part_idx_array_(const bool msg_all_part_merged_in, int ObElectionGroup::parse_majority_part_idx_array_(const bool msg_all_part_merged_in,
const ObPartIdxArray& msg_majority_part_idx_array, const ObPartitionArray& msg_part_array, const ObPartIdxArray &msg_majority_part_idx_array, const ObPartitionArray &msg_part_array,
ObPartIdxArray& majority_idx_array) const ObPartIdxArray &majority_idx_array) const
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
const int64_t msg_array_cnt = msg_part_array.count(); const int64_t msg_array_cnt = msg_part_array.count();
@ -1271,7 +1271,7 @@ int ObElectionGroup::parse_majority_part_idx_array_(const bool msg_all_part_merg
FORCE_ELECT_LOG(ERROR, "two array count not match", K(ret), "election_group", *this); FORCE_ELECT_LOG(ERROR, "two array count not match", K(ret), "election_group", *this);
} else if (msg_all_part_merged_in) { } else if (msg_all_part_merged_in) {
for (int64_t i = 0; OB_SUCC(ret) && i < msg_array_cnt; ++i) { for (int64_t i = 0; OB_SUCC(ret) && i < msg_array_cnt; ++i) {
const ObPartitionKey& tmp_pkey = msg_part_array.at(i); const ObPartitionKey &tmp_pkey = msg_part_array.at(i);
if (is_partition_exist_unlock_(tmp_pkey, tmp_idx) && OB_FAIL(majority_idx_array.push_back(tmp_idx))) { if (is_partition_exist_unlock_(tmp_pkey, tmp_idx) && OB_FAIL(majority_idx_array.push_back(tmp_idx))) {
FORCE_ELECT_LOG(WARN, "array push_back failed", K(ret)); FORCE_ELECT_LOG(WARN, "array push_back failed", K(ret));
} }
@ -1281,7 +1281,7 @@ int ObElectionGroup::parse_majority_part_idx_array_(const bool msg_all_part_merg
for (int64_t i = 0; OB_SUCC(ret) && i < msg_majority_array_cnt; ++i) { for (int64_t i = 0; OB_SUCC(ret) && i < msg_majority_array_cnt; ++i) {
int64_t cur_idx = msg_majority_part_idx_array.at(i); int64_t cur_idx = msg_majority_part_idx_array.at(i);
if (cur_idx >= 0 && cur_idx < msg_array_cnt) { if (cur_idx >= 0 && cur_idx < msg_array_cnt) {
const ObPartitionKey& tmp_pkey = msg_part_array.at(cur_idx); const ObPartitionKey &tmp_pkey = msg_part_array.at(cur_idx);
if (is_partition_exist_unlock_(tmp_pkey, tmp_idx) && OB_FAIL(majority_idx_array.push_back(tmp_idx))) { if (is_partition_exist_unlock_(tmp_pkey, tmp_idx) && OB_FAIL(majority_idx_array.push_back(tmp_idx))) {
FORCE_ELECT_LOG(WARN, "array push_back failed", K(ret)); FORCE_ELECT_LOG(WARN, "array push_back failed", K(ret));
} }
@ -1293,7 +1293,7 @@ int ObElectionGroup::parse_majority_part_idx_array_(const bool msg_all_part_merg
} }
// leader process vote_success message locally // leader process vote_success message locally
int ObElectionGroup::local_handle_vote_success_(const ObElectionMsgEGSuccess& msg) int ObElectionGroup::local_handle_vote_success_(const ObElectionMsgEGSuccess &msg)
{ {
// caller need add lock // caller need add lock
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -1344,7 +1344,7 @@ int ObElectionGroup::local_handle_vote_success_(const ObElectionMsgEGSuccess& ms
cur_ts - msg_t1); cur_ts - msg_t1);
} else { } else {
ObElectionInfo::StateHelper state_helper(state_); ObElectionInfo::StateHelper state_helper(state_);
const ObAddr& msg_new_leader = msg.get_new_leader(); const ObAddr &msg_new_leader = msg.get_new_leader();
const bool msg_all_part_merged_in = msg.is_all_part_merged_in(); const bool msg_all_part_merged_in = msg.is_all_part_merged_in();
// indicate should push election's entire lease or not // indicate should push election's entire lease or not
bool need_update_eg_lease = msg_all_part_merged_in; bool need_update_eg_lease = msg_all_part_merged_in;
@ -1386,9 +1386,9 @@ int ObElectionGroup::local_handle_vote_success_(const ObElectionMsgEGSuccess& ms
return ret; return ret;
} }
int ObElectionGroup::handle_vote_success(const ObElectionMsgEGSuccess& msg, const int64_t msg_eg_version, int ObElectionGroup::handle_vote_success(const ObElectionMsgEGSuccess &msg, const int64_t msg_eg_version,
const bool is_array_deserialized, const char* buf, const int64_t buf_len, const int64_t array_start_pos, const bool is_array_deserialized, const char *buf, const int64_t buf_len, const int64_t array_start_pos,
ObPartitionArray& msg_part_array, ObPartitionArray& move_in_failed_array, obrpc::ObElectionRpcResult& result) ObPartitionArray &msg_part_array, ObPartitionArray &move_in_failed_array, obrpc::ObElectionRpcResult &result)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -1444,7 +1444,7 @@ int ObElectionGroup::handle_vote_success(const ObElectionMsgEGSuccess& msg, cons
} }
if (OB_SUCC(ret)) { if (OB_SUCC(ret)) {
ObElectionInfo::StateHelper state_helper(state_); ObElectionInfo::StateHelper state_helper(state_);
const ObAddr& msg_new_leader = msg.get_new_leader(); const ObAddr &msg_new_leader = msg.get_new_leader();
const bool msg_all_part_merged_in = msg.is_all_part_merged_in(); const bool msg_all_part_merged_in = msg.is_all_part_merged_in();
bool need_update_merged_in_val = false; bool need_update_merged_in_val = false;
@ -1563,8 +1563,8 @@ int ObElectionGroup::handle_vote_success(const ObElectionMsgEGSuccess& msg, cons
return ret; return ret;
} }
int ObElectionGroup::leader_reappoint_(const bool need_update_eg_lease, const ObPartIdxArray& majority_part_idx_array, int ObElectionGroup::leader_reappoint_(const bool need_update_eg_lease, const ObPartIdxArray &majority_part_idx_array,
const ObAddr& leader, const int64_t t1, const int64_t lease_time) const ObAddr &leader, const int64_t t1, const int64_t lease_time)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
const int64_t cur_ts = get_current_ts(); const int64_t cur_ts = get_current_ts();
@ -1580,7 +1580,7 @@ int ObElectionGroup::leader_reappoint_(const bool need_update_eg_lease, const Ob
ObElectionRole::ROLE_LEADER == role_ && // need check lease only if I'm leader ObElectionRole::ROLE_LEADER == role_ && // need check lease only if I'm leader
cur_ts > leader_lease_.second) { // check if lease has been expired cur_ts > leader_lease_.second) { // check if lease has been expired
// maybe lease expired event has been exposed to others, leader epoch should pushed // maybe lease expired event has been exposed to others, leader epoch should pushed
// give up renew lease, group will be destoryed // give up renew lease, group will be destroyed
ret = OB_ELECTION_WARN_LEADER_LEASE_EXPIRED; ret = OB_ELECTION_WARN_LEADER_LEASE_EXPIRED;
FORCE_ELECT_LOG(WARN, FORCE_ELECT_LOG(WARN,
"group leader lease expired, give up reappointing", "group leader lease expired, give up reappointing",
@ -1621,7 +1621,7 @@ int ObElectionGroup::leader_reappoint_(const bool need_update_eg_lease, const Ob
return ret; return ret;
} }
int ObElectionGroup::leader_elected_(const ObAddr& leader, const int64_t t1, const int64_t lease_time) int ObElectionGroup::leader_elected_(const ObAddr &leader, const int64_t t1, const int64_t lease_time)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -1667,7 +1667,7 @@ bool ObElectionGroup::lease_expired_too_long_(const int64_t cur_ts) const
} }
int ObElectionGroup::handle_prepare_destroy_msg( int ObElectionGroup::handle_prepare_destroy_msg(
const ObElectionMsg& msg, const int64_t msg_eg_version, obrpc::ObElectionRpcResult& result) const ObElectionMsg &msg, const int64_t msg_eg_version, obrpc::ObElectionRpcResult &result)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -1748,7 +1748,7 @@ int ObElectionGroup::prepare_destroy()
return ret; return ret;
} }
int ObElectionGroup::get_priority_(ObElectionGroupPriority& priority) const int ObElectionGroup::get_priority_(ObElectionGroupPriority &priority) const
{ {
// only leader group need call this to get priority for now // only leader group need call this to get priority for now
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -2020,7 +2020,7 @@ bool ObElectionGroup::vote_run_time_out_of_range_(const int64_t cur_ts, const in
return (cur_ts - expect_ts) > T_VOTE_TIMER_DIFF; return (cur_ts - expect_ts) > T_VOTE_TIMER_DIFF;
} }
int ObElectionGroup::send_vote_prepare_(const ObElectionMsgEGPrepare& msg) int ObElectionGroup::send_vote_prepare_(const ObElectionMsgEGPrepare &msg)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -2046,7 +2046,7 @@ int ObElectionGroup::send_vote_prepare_(const ObElectionMsgEGPrepare& msg)
return ret; return ret;
} }
int ObElectionGroup::send_vote_success_(const ObElectionMsgEGSuccess& msg) int ObElectionGroup::send_vote_success_(const ObElectionMsgEGSuccess &msg)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
const int64_t cur_ts = get_current_ts(); const int64_t cur_ts = get_current_ts();
@ -2065,7 +2065,7 @@ int ObElectionGroup::send_vote_success_(const ObElectionMsgEGSuccess& msg)
return ret; return ret;
} }
int ObElectionGroup::get_eg_centralized_candidate_(ObAddr& cur_leader, ObAddr& new_leader) int ObElectionGroup::get_eg_centralized_candidate_(ObAddr &cur_leader, ObAddr &new_leader)
{ {
int ret = msg_pool_.get_eg_centralized_candidate(cur_leader, new_leader, T1_timestamp_); int ret = msg_pool_.get_eg_centralized_candidate(cur_leader, new_leader, T1_timestamp_);
if (OB_FAIL(ret)) { if (OB_FAIL(ret)) {
@ -2075,7 +2075,7 @@ int ObElectionGroup::get_eg_centralized_candidate_(ObAddr& cur_leader, ObAddr& n
return ret; return ret;
} }
int ObElectionGroup::get_unconfirmed_leader_info(ObAddr& unconfirmed_leader, lease_t& unconfirmed_leader_lease) const int ObElectionGroup::get_unconfirmed_leader_info(ObAddr &unconfirmed_leader, lease_t &unconfirmed_leader_lease) const
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -2086,8 +2086,8 @@ int ObElectionGroup::get_unconfirmed_leader_info(ObAddr& unconfirmed_leader, lea
return ret; return ret;
} }
int ObElectionGroup::get_leader_lease_info(const ObPartitionKey& pkey, const int64_t target_idx, int ObElectionGroup::get_leader_lease_info(const ObPartitionKey &pkey, const int64_t target_idx,
const int64_t part_cur_ts, int64_t& eg_lease_end, int64_t& eg_takeover_t1_ts) const const int64_t part_cur_ts, int64_t &eg_lease_end, int64_t &eg_takeover_t1_ts) const
{ {
// called from election // called from election
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -2131,7 +2131,7 @@ int ObElectionGroup::get_leader_lease_info(const ObPartitionKey& pkey, const int
} }
bool ObElectionGroup::part_leader_lease_is_expired( bool ObElectionGroup::part_leader_lease_is_expired(
const int64_t part_cur_ts, const ObPartitionKey& pkey, const int64_t target_idx, int64_t& out_lease_end) const const int64_t part_cur_ts, const ObPartitionKey &pkey, const int64_t target_idx, int64_t &out_lease_end) const
{ {
// called from election, checking if lease has been expired // called from election, checking if lease has been expired
RLockGuard guard(lock_); RLockGuard guard(lock_);
@ -2139,7 +2139,7 @@ bool ObElectionGroup::part_leader_lease_is_expired(
} }
bool ObElectionGroup::part_leader_lease_is_expired_unlock_( bool ObElectionGroup::part_leader_lease_is_expired_unlock_(
const int64_t part_cur_ts, const ObPartitionKey& pkey, const int64_t target_idx, int64_t& out_lease_end) const const int64_t part_cur_ts, const ObPartitionKey &pkey, const int64_t target_idx, int64_t &out_lease_end) const
{ {
bool bool_ret = true; bool bool_ret = true;
const int64_t cur_ts = get_current_ts(); const int64_t cur_ts = get_current_ts();
@ -2182,7 +2182,7 @@ bool ObElectionGroup::part_leader_lease_is_expired_unlock_(
} }
int ObElectionGroup::check_all_part_leader_lease_( int ObElectionGroup::check_all_part_leader_lease_(
const int64_t cur_ts, bool& is_all_expired, ObPartitionArray& expired_part_array) const const int64_t cur_ts, bool &is_all_expired, ObPartitionArray &expired_part_array) const
{ {
// check all partitions' lease in group // check all partitions' lease in group
// will be called when is_all_part_merged_in_=false // will be called when is_all_part_merged_in_=false
@ -2202,7 +2202,7 @@ int ObElectionGroup::check_all_part_leader_lease_(
// group is not empty, check every partition's lease // group is not empty, check every partition's lease
const int64_t part_array_cnt = partition_array_.count(); const int64_t part_array_cnt = partition_array_.count();
for (int64_t i = 0; OB_SUCC(ret) && i < part_array_cnt; ++i) { for (int64_t i = 0; OB_SUCC(ret) && i < part_array_cnt; ++i) {
const ObPartitionKey& tmp_pkey = partition_array_.at(i); const ObPartitionKey &tmp_pkey = partition_array_.at(i);
const int64_t tmp_lease_end = part_state_array_.at(i).lease_end_; const int64_t tmp_lease_end = part_state_array_.at(i).lease_end_;
if (cur_ts <= tmp_lease_end) { if (cur_ts <= tmp_lease_end) {
// not expired // not expired
@ -2216,7 +2216,7 @@ int ObElectionGroup::check_all_part_leader_lease_(
return ret; return ret;
} }
bool ObElectionGroup::eg_lease_is_completely_expired_(const int64_t cur_ts, bool& is_all_dropped) const bool ObElectionGroup::eg_lease_is_completely_expired_(const int64_t cur_ts, bool &is_all_dropped) const
{ {
bool bool_ret = true; bool bool_ret = true;
@ -2237,7 +2237,7 @@ bool ObElectionGroup::eg_lease_is_completely_expired_(const int64_t cur_ts, bool
is_all_dropped = true; is_all_dropped = true;
const int64_t part_array_cnt = expired_part_array.count(); const int64_t part_array_cnt = expired_part_array.count();
for (int64_t i = 0; is_all_dropped && i < part_array_cnt; ++i) { for (int64_t i = 0; is_all_dropped && i < part_array_cnt; ++i) {
const ObPartitionKey& tmp_pkey = expired_part_array.at(i); const ObPartitionKey &tmp_pkey = expired_part_array.at(i);
bool is_exist = true; bool is_exist = true;
if (OB_SUCCESS != if (OB_SUCCESS !=
(tmp_ret = storage::ObPartitionService::get_instance().check_partition_exist(tmp_pkey, is_exist))) { (tmp_ret = storage::ObPartitionService::get_instance().check_partition_exist(tmp_pkey, is_exist))) {
@ -2278,12 +2278,12 @@ bool ObElectionGroup::unconfirmed_leader_lease_is_expired_(const int64_t cur_ts)
return (cur_ts > unconfirmed_leader_lease_.second); return (cur_ts > unconfirmed_leader_lease_.second);
} }
void ObElectionGroup::set_unconfirmed_leader_(const ObAddr& unconfirmed_leader) void ObElectionGroup::set_unconfirmed_leader_(const ObAddr &unconfirmed_leader)
{ {
unconfirmed_leader_ = unconfirmed_leader; unconfirmed_leader_ = unconfirmed_leader;
} }
int ObElectionGroup::post_election_msg_(const ObMemberList& mlist, const ObElectionMsg& msg) int ObElectionGroup::post_election_msg_(const ObMemberList &mlist, const ObElectionMsg &msg)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
ObAddr server; ObAddr server;
@ -2325,7 +2325,7 @@ int ObElectionGroup::check_array_buf_()
return ret; return ret;
} }
int ObElectionGroup::post_election_msg_(const ObAddr& server, const ObElectionMsg& msg) int ObElectionGroup::post_election_msg_(const ObAddr &server, const ObElectionMsg &msg)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -2343,12 +2343,12 @@ int ObElectionGroup::post_election_msg_(const ObAddr& server, const ObElectionMs
return ret; return ret;
} }
bool ObElectionGroup::is_candidate_(const ObAddr& server) const bool ObElectionGroup::is_candidate_(const ObAddr &server) const
{ {
return sorted_member_list_.contains(server); return sorted_member_list_.contains(server);
} }
bool ObElectionGroup::is_real_leader_(const ObAddr& server) const bool ObElectionGroup::is_real_leader_(const ObAddr &server) const
{ {
bool bool_ret = false; bool bool_ret = false;
const int64_t cur_ts = get_current_ts(); const int64_t cur_ts = get_current_ts();
@ -2378,7 +2378,7 @@ bool ObElectionGroup::in_reappoint_period_(const int64_t ts)
return bool_ret; return bool_ret;
} }
int ObElectionGroup::get_election_group_info(ObElectionGroupInfo& eg_info) const int ObElectionGroup::get_election_group_info(ObElectionGroupInfo &eg_info) const
{ {
RLockGuard guard(lock_); RLockGuard guard(lock_);
eg_info.set_is_running(is_running_); eg_info.set_is_running(is_running_);

38
src/election/ob_election_group_msg_pool.cpp
View File

@ -25,7 +25,7 @@ void ObEGVoteMsgPool::reset()
} }
template <typename T> template <typename T>
int ObEGVoteMsgPool::do_store_(const T& msg, ObElectionVoteMsgArray<T>& array) int ObEGVoteMsgPool::do_store_(const T &msg, ObElectionVoteMsgArray<T> &array)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
bool is_exist = false; bool is_exist = false;
@ -59,9 +59,9 @@ int ObEGVoteMsgPool::do_store_(const T& msg, ObElectionVoteMsgArray<T>& array)
return ret; return ret;
} }
int ObEGVoteMsgPool::check_eg_centralized_majority(common::ObAddr& cur_leader, common::ObAddr& new_leader, int ObEGVoteMsgPool::check_eg_centralized_majority(common::ObAddr &cur_leader, common::ObAddr &new_leader,
const bool is_all_part_merged_in, bool& is_eg_majority, ObPartStateArray& part_state_array, const bool is_all_part_merged_in, bool &is_eg_majority, ObPartStateArray &part_state_array,
const common::ObPartitionArray& partition_array, const int64_t self_version, const int64_t replica_num, const common::ObPartitionArray &partition_array, const int64_t self_version, const int64_t replica_num,
const int64_t t1, const uint64_t eg_hash, const int64_t cur_ts) const int64_t t1, const uint64_t eg_hash, const int64_t cur_ts)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -106,7 +106,7 @@ int ObEGVoteMsgPool::check_eg_centralized_majority(common::ObAddr& cur_leader, c
return ret; return ret;
} }
int ObEGVoteMsgPool::store(const ObElectionMsg& msg) int ObEGVoteMsgPool::store(const ObElectionMsg &msg)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -116,9 +116,9 @@ int ObEGVoteMsgPool::store(const ObElectionMsg& msg)
} else { } else {
const int msg_type = msg.get_msg_type(); const int msg_type = msg.get_msg_type();
if (msg_type == OB_ELECTION_EG_VOTE_PREPARE) { if (msg_type == OB_ELECTION_EG_VOTE_PREPARE) {
ret = store_(static_cast<const ObElectionMsgEGPrepare&>(msg)); ret = store_(static_cast<const ObElectionMsgEGPrepare &>(msg));
} else if (msg_type == OB_ELECTION_EG_VOTE_VOTE) { } else if (msg_type == OB_ELECTION_EG_VOTE_VOTE) {
ret = store_(static_cast<const ObElectionMsgEGVote4Store&>(msg)); ret = store_(static_cast<const ObElectionMsgEGVote4Store &>(msg));
} else { } else {
ret = OB_ERR_UNEXPECTED; ret = OB_ERR_UNEXPECTED;
} }
@ -127,7 +127,7 @@ int ObEGVoteMsgPool::store(const ObElectionMsg& msg)
return ret; return ret;
} }
int ObEGVoteMsgPool::store_(const ObElectionMsgEGPrepare& msg) int ObEGVoteMsgPool::store_(const ObElectionMsgEGPrepare &msg)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -142,7 +142,7 @@ int ObEGVoteMsgPool::store_(const ObElectionMsgEGPrepare& msg)
return ret; return ret;
} }
int ObEGVoteMsgPool::store_(const ObElectionMsgEGVote4Store& msg) int ObEGVoteMsgPool::store_(const ObElectionMsgEGVote4Store &msg)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -157,7 +157,7 @@ int ObEGVoteMsgPool::store_(const ObElectionMsgEGVote4Store& msg)
return ret; return ret;
} }
int ObEGVoteMsgPool::get_eg_centralized_candidate(ObAddr& cur_leader, ObAddr& new_leader, const int64_t t1) int ObEGVoteMsgPool::get_eg_centralized_candidate(ObAddr &cur_leader, ObAddr &new_leader, const int64_t t1)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
int count = 0; int count = 0;
@ -199,7 +199,7 @@ int ObEGVoteMsgPool::get_eg_centralized_candidate(ObAddr& cur_leader, ObAddr& ne
} }
bool ObEGVoteMsgPool::is_pkey_exist_( bool ObEGVoteMsgPool::is_pkey_exist_(
const common::ObPartitionKey& pkey, const common::ObPartitionArray& partition_array, int64_t& idx) const const common::ObPartitionKey &pkey, const common::ObPartitionArray &partition_array, int64_t &idx) const
{ {
bool bool_ret = false; bool bool_ret = false;
idx = -1; idx = -1;
@ -213,8 +213,8 @@ bool ObEGVoteMsgPool::is_pkey_exist_(
return bool_ret; return bool_ret;
} }
int ObEGVoteMsgPool::count_ticket_(const common::ObPartitionArray& msg_part_array, int ObEGVoteMsgPool::count_ticket_(const common::ObPartitionArray &msg_part_array,
const common::ObPartitionArray& partition_array, ObPartStateArray& part_state_array) const common::ObPartitionArray &partition_array, ObPartStateArray &part_state_array)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -236,7 +236,7 @@ int ObEGVoteMsgPool::count_ticket_(const common::ObPartitionArray& msg_part_arra
} }
int ObEGVoteMsgPool::process_vote_cnt_(const int64_t eg_vote_cnt, const int64_t replica_num, int ObEGVoteMsgPool::process_vote_cnt_(const int64_t eg_vote_cnt, const int64_t replica_num,
ObPartStateArray& part_state_array, bool& is_eg_majority, const int64_t cur_ts) ObPartStateArray &part_state_array, bool &is_eg_majority, const int64_t cur_ts)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -257,9 +257,9 @@ int ObEGVoteMsgPool::process_vote_cnt_(const int64_t eg_vote_cnt, const int64_t
return ret; return ret;
} }
int ObEGVoteMsgPool::get_eg_centralized_majority_(common::ObAddr& cur_leader, common::ObAddr& new_leader, int ObEGVoteMsgPool::get_eg_centralized_majority_(common::ObAddr &cur_leader, common::ObAddr &new_leader,
const bool is_all_part_merged_in, bool& is_eg_majority, ObPartStateArray& part_state_array, const bool is_all_part_merged_in, bool &is_eg_majority, ObPartStateArray &part_state_array,
const common::ObPartitionArray& partition_array, const int64_t self_version, const int64_t replica_num, const common::ObPartitionArray &partition_array, const int64_t self_version, const int64_t replica_num,
const int64_t t1, const uint64_t hash, const int64_t cur_ts) const int64_t t1, const uint64_t hash, const int64_t cur_ts)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -276,7 +276,7 @@ int ObEGVoteMsgPool::get_eg_centralized_majority_(common::ObAddr& cur_leader, co
ELECT_ASYNC_LOG(WARN, "message t1 timestamp not match", K(i), K(hash), K(t1), K(msg), K(msg_cnt)); ELECT_ASYNC_LOG(WARN, "message t1 timestamp not match", K(i), K(hash), K(t1), K(msg), K(msg_cnt));
// yes, continue // yes, continue
} else { } else {
int64_t ticket = 0; // totle ticket int64_t ticket = 0; // total ticket
int64_t eg_vote_cnt = 0; // the ticket in same version int64_t eg_vote_cnt = 0; // the ticket in same version
bool is_leader_sender = false; bool is_leader_sender = false;
tmp_cur_leader = msg.get_cur_leader(); tmp_cur_leader = msg.get_cur_leader();
@ -313,7 +313,7 @@ int ObEGVoteMsgPool::get_eg_centralized_majority_(common::ObAddr& cur_leader, co
if (eg_vote_cnt > replica_num / 2) { if (eg_vote_cnt > replica_num / 2) {
if (false == is_all_part_merged_in) { // first merge-in, need check lease of every election if (false == is_all_part_merged_in) { // first merge-in, need check lease of every election
process_vote_cnt_(eg_vote_cnt, replica_num, part_state_array, is_eg_majority, cur_ts); process_vote_cnt_(eg_vote_cnt, replica_num, part_state_array, is_eg_majority, cur_ts);
} else { // already in merge-in state, no need to check lease of singal election, cause that lease is not } else { // already in merge-in state, no need to check lease of single election, cause that lease is not
// maintained anymore // maintained anymore
is_eg_majority = true; is_eg_majority = true;
} }

2287
src/rootserver/ob_ddl_service.cpp
View File

File diff suppressed because it is too large Load Diff