* Copyright (c) 2021 Huawei Technologies Co.,Ltd.
*
* openGauss is licensed under Mulan PSL v2.
* You can use this software according to the terms and conditions of the Mulan PSL v2.
* You may obtain a copy of Mulan PSL v2 at:
*
* http://license.coscl.org.cn/MulanPSL2
*
* 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 PSL v2 for more details.
* -------------------------------------------------------------------------
*
* elc_stream.c
* election process
*
* IDENTIFICATION
* src/election/elc_stream.c
*
* -------------------------------------------------------------------------
*/
#include "elc_stream.h"
#include "stg_manager.h"
#include "mec.h"
#include "cm_timer.h"
#include "replication.h"
#include "util_defs.h"
#include "elc_status_check.h"
#ifdef __cplusplus
extern "C" {
#endif
typedef struct st_elc_info {
latch_t latch;
uint32 stream_id;
uint64 current_term;
uint32 votefor_id;
uint32 vote_count;
uint32 vote_no_count;
dcf_role_t node_role;
timespec_t last_hb_time;
timespec_t last_hb_ack[CM_MAX_NODE_COUNT];
dcf_work_mode_t work_mode;
uint32 vote_num;
dcf_work_mode_t vote_node_work_mode[CM_MAX_NODE_COUNT];
uint32 old_leader_id;
timespec_t last_md_rep_time;
volatile bool32 force_vote;
bool32 inter_promote_flag;
volatile timespec_t leader_start_time;
uint32 leader_group;
uint32 my_group;
uint64 priority;
} elc_info_t;
static elc_info_t g_stream_list[CM_MAX_STREAM_COUNT];
static role_notify_t g_stream_notify[CM_MAX_STREAM_COUNT];
cm_thread_cond_t g_status_notify_cond;
usr_cb_status_notify_t g_cb_status_notify = NULL;
usr_cb_election_notify_t g_cb_election_notify = NULL;
status_t get_current_node_role(uint32 stream_id, uint32 current_node_id, uint32 leader_id, dcf_role_t *role)
{
dcf_node_t node_info;
CM_RETURN_IFERR(md_get_stream_node_ext(stream_id, current_node_id, &node_info));
uint32 node_num;
CM_RETURN_IFERR(md_get_stream_nodes_count(stream_id, &node_num));
if (node_num == 1) {
if (node_info.default_role != DCF_ROLE_LEADER) {
LOG_RUN_WAR("[ELC] 1 node mode, force default_role(%d) to leader", node_info.default_role);
}
*role = DCF_ROLE_LEADER;
(void)elc_stream_set_votefor(stream_id, current_node_id);
return CM_SUCCESS;
}
param_run_mode_t run_mode = elc_stream_get_run_mode();
if (run_mode == ELECTION_AUTO || run_mode == ELECTION_DISABLE) {
if (node_info.default_role == DCF_ROLE_LEADER) {
*role = DCF_ROLE_FOLLOWER;
} else {
*role = node_info.default_role;
}
} else {
if (leader_id == CM_INVALID_NODE_ID) {
*role = node_info.default_role;
} else {
if (current_node_id == leader_id) {
*role = DCF_ROLE_LEADER;
} else {
*role = (node_info.default_role == DCF_ROLE_LEADER) ? DCF_ROLE_FOLLOWER : node_info.default_role;
}
}
}
return CM_SUCCESS;
}
uint32 elc_stream_get_elc_timeout_ms()
{
param_value_t value;
if (md_get_param(DCF_PARAM_ELECTION_TIMEOUT, &value) == CM_SUCCESS) {
return value.value_elc_timeout;
} else {
return (uint32)CM_DEFAULT_ELC_TIMEOUT;
}
}
uint32 elc_stream_get_hb_interval_ms()
{
param_value_t value;
if (md_get_param(DCF_PARAM_HEARTBEAT_INTERVAL, &value) == CM_SUCCESS) {
return value.value_hb_interval;
} else {
return (uint32)CM_DEFAULT_HB_INTERVAL;
}
}
uint32 elc_stream_get_elc_switch_thd_sec()
{
param_value_t value;
if (md_get_param(DCF_PARAM_ELECTION_SWITCH_THRESHOLD, &value) == CM_SUCCESS) {
return value.value_elc_switch_thd;
} else {
return (uint32)CM_DEFAULT_ELC_SWITCH_THD;
}
}
bool32 elc_stream_get_auto_elc_pri_en()
{
param_value_t value;
if (md_get_param(DCF_PARAM_AUTO_ELC_PRIORITY_EN, &value) == CM_SUCCESS) {
return value.value_auto_elc_priority_en;
} else {
return (bool32)CM_TRUE;
}
}
param_run_mode_t elc_stream_get_run_mode()
{
param_value_t value;
if (md_get_param(DCF_PARAM_RUN_MODE, &value) == CM_SUCCESS) {
return value.value_mode;
} else {
return ELECTION_MANUAL;
}
}
status_t elc_stream_init()
{
cm_init_cond(&g_status_notify_cond);
MEMS_RETURN_IFERR(memset_sp(g_stream_notify, sizeof(role_notify_t) * CM_MAX_STREAM_COUNT, 0,
sizeof(role_notify_t) * CM_MAX_STREAM_COUNT));
MEMS_RETURN_IFERR(memset_sp(g_stream_list, sizeof(elc_info_t) * CM_MAX_STREAM_COUNT, 0,
sizeof(elc_info_t) * CM_MAX_STREAM_COUNT));
uint32 current_node_id = md_get_cur_node();
uint32 stream_list[CM_MAX_STREAM_COUNT];
uint32 stream_count;
CM_RETURN_IFERR(md_get_stream_list(stream_list, &stream_count));
for (uint32 i = 0; i < stream_count; i++) {
uint32 stream_id = stream_list[i];
uint32 leader_id = stg_get_votedfor(stream_id);
g_stream_list[stream_id].stream_id = stream_id;
g_stream_list[stream_id].votefor_id = leader_id;
g_stream_list[stream_id].current_term = stg_get_current_term(stream_id);
g_stream_list[stream_id].vote_count = 0;
g_stream_list[stream_id].vote_no_count = 0;
dcf_role_t node_role;
CM_RETURN_IFERR(get_current_node_role(stream_id, current_node_id, leader_id, &node_role));
g_stream_list[stream_id].node_role = node_role;
g_stream_list[stream_id].vote_num = 0;
g_stream_list[stream_id].old_leader_id = CM_INVALID_ID32;
g_stream_list[stream_id].work_mode = WM_NORMAL;
g_stream_list[stream_id].force_vote = CM_FALSE;
g_stream_list[stream_id].inter_promote_flag = CM_FALSE;
dcf_node_t node_info;
CM_RETURN_IFERR(md_get_stream_node_ext(stream_id, current_node_id, &node_info));
g_stream_list[stream_id].my_group = node_info.group;
g_stream_list[stream_id].leader_group = node_info.group;
g_stream_list[stream_id].priority = node_info.priority;
if ((leader_id == current_node_id) || (leader_id == CM_INVALID_NODE_ID
&& node_info.default_role == DCF_ROLE_LEADER)) {
g_stream_list[stream_id].last_hb_time = 0;
} else {
g_stream_list[stream_id].last_hb_time =
cm_clock_now() + CM_MAX_ELC_INIT_WAIT_TIMES * elc_stream_get_elc_timeout_ms() * MICROSECS_PER_MILLISEC;
}
LOG_RUN_INF("[ELC]stream %u init, cur_node_id %u, vote_for %u, last_hb_time %lld",
stream_id, current_node_id, leader_id, g_stream_list[stream_id].last_hb_time);
for (uint32 j = 0; j < CM_MAX_NODE_COUNT; j++) {
g_stream_list[stream_id].last_hb_ack[j] = cm_clock_now();
}
cm_latch_init(&g_stream_list[stream_id].latch);
}
return CM_SUCCESS;
}
uint64 elc_stream_get_current_term(uint32 stream_id)
{
return g_stream_list[stream_id].current_term;
}
status_t elc_stream_set_term(uint32 stream_id, uint64 current_term)
{
g_stream_list[stream_id].current_term = current_term;
return stg_set_current_term(stream_id, current_term);
}
dcf_role_t elc_stream_get_role(uint32 stream_id)
{
return g_stream_list[stream_id].node_role;
}
void add_notify_item(uint32 stream_id, uint32 node_id, uint32 new_leader, dcf_role_t old_role, dcf_role_t new_role)
{
uint32 count = 0;
LOG_DEBUG_INF("[ELC]add_notify_item start");
cm_latch_x(&g_stream_notify[stream_id].latch, 0, NULL);
do {
uint32 pi = g_stream_notify[stream_id].pi;
if ((pi + 1) % MAX_NOTIFY_ITEM_NUM != g_stream_notify[stream_id].ci) {
g_stream_notify[stream_id].item[pi].node_id = node_id;
g_stream_notify[stream_id].item[pi].new_leader = new_leader;
g_stream_notify[stream_id].item[pi].old_role = old_role;
g_stream_notify[stream_id].item[pi].new_role = new_role;
LOG_DEBUG_INF("[ELC]added item, pi=%u, stream_id=%u, node_id=%u new_leader=%u old_role=%d new_role=%d",
pi, stream_id, node_id, new_leader, old_role, new_role);
g_stream_notify[stream_id].pi = (pi + 1) % MAX_NOTIFY_ITEM_NUM;
break;
} else {
cm_unlatch(&g_stream_notify[stream_id].latch, NULL);
cm_sleep(CM_SLEEP_10_FIXED);
count++;
if (count > CM_100X_FIXED) {
LOG_RUN_ERR("[ELC]add_item timeout.stream_id=%u, node_id=%u new_leader=%u old_role=%d new_role=%d",
stream_id, node_id, new_leader, old_role, new_role);
return;
}
cm_latch_x(&g_stream_notify[stream_id].latch, 0, NULL);
}
} while (1);
cm_unlatch(&g_stream_notify[stream_id].latch, NULL);
cm_release_cond(&g_status_notify_cond);
LOG_DEBUG_INF("[ELC]add_notify_item end");
}
status_t get_notify_item(uint32 stream_id, role_notify_item_t* notify_item)
{
cm_latch_x(&g_stream_notify[stream_id].latch, 0, NULL);
uint32 ci = g_stream_notify[stream_id].ci;
if (ci != g_stream_notify[stream_id].pi) {
*notify_item = g_stream_notify[stream_id].item[ci];
g_stream_notify[stream_id].ci = (ci + 1) % MAX_NOTIFY_ITEM_NUM;
cm_unlatch(&g_stream_notify[stream_id].latch, NULL);
LOG_DEBUG_INF("[ELC]get item success, ci=%u, stream_id=%u", ci, stream_id);
return CM_SUCCESS;
}
cm_unlatch(&g_stream_notify[stream_id].latch, NULL);
return CM_ERROR;
}
void rep_set_can_write_flag(uint32 stream_id, uint32 flag);
void elc_stream_notify_proc()
{
(void)cm_wait_cond(&g_status_notify_cond, CM_SLEEP_500_FIXED);
role_notify_item_t notify_item;
for (uint32 i = 0; i < CM_MAX_STREAM_COUNT; i++) {
uint32 stream_id = i;
if (get_notify_item(stream_id, ¬ify_item) == CM_SUCCESS) {
dcf_role_t old_role = notify_item.old_role;
dcf_role_t role = notify_item.new_role;
LOG_DEBUG_INF("[ELC]got notify item: stream_id=%u old_role=%d new_role=%d", stream_id, old_role, role);
if ((old_role != DCF_ROLE_LEADER && role == DCF_ROLE_LEADER) ||
(old_role == DCF_ROLE_LEADER && role != DCF_ROLE_LEADER)) {
LOG_DEBUG_INF("[ELC]status_changed_notify proc begin");
rep_set_can_write_flag(stream_id, CM_FALSE);
uint32 prio_leader = elc_get_rcv_best_priority_node(stream_id);
bool32 force_vote = elc_stream_is_force_vote(stream_id);
elc_stream_set_force_vote_flag(stream_id, CM_FALSE);
LOG_RUN_INF("[ELC]max_prio_leader=%u force_vote=%d role=%d", prio_leader, force_vote, role);
if (force_vote == CM_FALSE && role == DCF_ROLE_LEADER && prio_leader != CM_INVALID_NODE_ID) {
rep_try_promote_prio_leader(stream_id, prio_leader);
if (elc_stream_get_role(stream_id) != DCF_ROLE_LEADER) {
elc_stream_set_inter_promote_flag(stream_id, CM_TRUE);
continue;
}
}
if (elc_stream_get_inter_promote_flag(stream_id) == CM_TRUE && role != DCF_ROLE_LEADER) {
continue;
}
elc_stream_set_inter_promote_flag(stream_id, CM_FALSE);
(void)rep_role_notify(stream_id, old_role, role);
if (g_cb_status_notify != NULL) {
int ret = g_cb_status_notify(stream_id,
(uint32)((role == DCF_ROLE_LEADER) ? DCF_ROLE_LEADER : DCF_ROLE_FOLLOWER));
LOG_DEBUG_INF("[ELC]Callback: status_changed_notify, retcode=%d", ret);
}
}
if (notify_item.new_leader != CM_INVALID_NODE_ID) {
if (g_cb_election_notify != NULL) {
int ret = g_cb_election_notify(stream_id, notify_item.new_leader);
LOG_DEBUG_INF("[ELC]Callback: status_election_notify, retcode=%d", ret);
}
notify_item.new_leader = CM_INVALID_NODE_ID;
}
}
}
}
status_t elc_stream_set_role(uint32 stream_id, dcf_role_t role)
{
dcf_role_t old_role = g_stream_list[stream_id].node_role;
g_stream_list[stream_id].node_role = role;
LOG_DEBUG_INF("[ELC]elc stream set role, stream_id=%u old_role=%d new_role=%d ", stream_id, old_role, role);
if ((old_role != DCF_ROLE_LEADER && role == DCF_ROLE_LEADER) ||
(old_role == DCF_ROLE_LEADER && role != DCF_ROLE_LEADER)) {
add_notify_item(stream_id, md_get_cur_node(), CM_INVALID_NODE_ID, old_role, role);
}
return CM_SUCCESS;
}
uint32 elc_stream_get_vote_count(uint32 stream_id)
{
return g_stream_list[stream_id].vote_count;
}
uint32 elc_stream_get_vote_no_count(uint32 stream_id)
{
return g_stream_list[stream_id].vote_no_count;
}
uint32 elc_stream_is_win(uint32 stream_id, bool32* is_win)
{
uint32 quorum;
CM_RETURN_IFERR(elc_stream_get_quorum(stream_id, &quorum));
*is_win = g_stream_list[stream_id].vote_count >= quorum;
return CM_SUCCESS;
}
status_t elc_stream_is_not_win(uint32 stream_id, bool32* is_not_win)
{
uint32 voter_num = 0;
CM_RETURN_IFERR(md_get_voter_num(stream_id, &voter_num));
if (g_stream_list[stream_id].work_mode == WM_NORMAL) {
*is_not_win = g_stream_list[stream_id].vote_no_count >= ((voter_num + 1) / CM_2X_FIXED);
} else if (g_stream_list[stream_id].work_mode == WM_MINORITY) {
*is_not_win = g_stream_list[stream_id].vote_no_count > (voter_num - g_stream_list[stream_id].vote_num);
} else {
LOG_RUN_ERR("invalid work_mode:%d", g_stream_list[stream_id].work_mode);
return CM_ERROR;
}
return CM_SUCCESS;
}
status_t elc_stream_get_quorum(uint32 stream_id, uint32* quorum)
{
uint32 voter_num = 0;
if (g_stream_list[stream_id].work_mode == WM_NORMAL) {
CM_RETURN_IFERR(md_get_voter_num(stream_id, &voter_num));
*quorum = (voter_num / CM_2X_FIXED) + 1;
} else if (g_stream_list[stream_id].work_mode == WM_MINORITY) {
*quorum = g_stream_list[stream_id].vote_num;
} else {
LOG_RUN_ERR("invalid work_mode:%d", g_stream_list[stream_id].work_mode);
return CM_ERROR;
}
return CM_SUCCESS;
}
status_t elc_stream_increase_vote_count(uint32 stream_id, uint32 voting_weight)
{
g_stream_list[stream_id].vote_count += voting_weight;
return CM_SUCCESS;
}
status_t elc_stream_increase_vote_no_count(uint32 stream_id, uint32 voting_weight)
{
g_stream_list[stream_id].vote_no_count += voting_weight;
return CM_SUCCESS;
}
void elc_stream_reset_vote_count(uint32 stream_id)
{
g_stream_list[stream_id].vote_count = 0;
g_stream_list[stream_id].vote_no_count = 0;
}
timespec_t elc_stream_get_timeout(uint32 stream_id)
{
return g_stream_list[stream_id].last_hb_time;
}
status_t elc_stream_set_timeout(uint32 stream_id, timespec_t time)
{
g_stream_list[stream_id].last_hb_time = time;
return CM_SUCCESS;
}
bool32 elc_stream_is_future_hb(uint32 stream_id)
{
uint64 hb_time = g_stream_list[stream_id].last_hb_time;
return (hb_time > cm_clock_now());
}
timespec_t elc_stream_get_hb_ack_time(uint32 stream_id, uint32 node_id)
{
return g_stream_list[stream_id].last_hb_ack[node_id];
}
status_t elc_stream_set_hb_ack_time(uint32 stream_id, uint32 node_id, timespec_t time)
{
g_stream_list[stream_id].last_hb_ack[node_id] = time;
return CM_SUCCESS;
}
uint32 elc_stream_get_votefor(uint32 stream_id)
{
return g_stream_list[stream_id].votefor_id;
}
status_t elc_stream_set_votefor(uint32 stream_id, uint32 votefor_id)
{
LOG_DEBUG_INF("[ELC]set votefor_id to %u", votefor_id);
if (votefor_id != CM_INVALID_NODE_ID && votefor_id != md_get_cur_node()) {
timespec_t now = cm_clock_now();
elc_stream_set_leader_start_time(stream_id, now);
LOG_DEBUG_INF("[ELC]set leader_start_time to %llu, votefor_id=%u", now, votefor_id);
}
g_stream_list[stream_id].votefor_id = votefor_id;
return stg_set_votedfor(stream_id, votefor_id);
}
uint32 elc_stream_get_old_leader(uint32 stream_id)
{
return g_stream_list[stream_id].old_leader_id;
}
uint32 elc_stream_set_old_leader(uint32 stream_id, uint32 leader_id)
{
g_stream_list[stream_id].old_leader_id = leader_id;
return CM_SUCCESS;
}
status_t elc_stream_vote_node_list(uint32 stream_id, uint64* inst_bits)
{
uint32 node_list[CM_MAX_NODE_COUNT];
uint32 node_count;
CM_RETURN_IFERR(md_get_stream_nodes(stream_id, node_list, &node_count));
uint32 current_node_id = elc_stream_get_current_node();
for (uint32 i = 0; i < node_count; i++) {
uint32 id = node_list[i];
if (id != current_node_id) {
MEC_SET_BRD_INST(inst_bits, id);
}
}
return CM_SUCCESS;
}
bool32 elc_stream_is_exists(uint32 stream_id)
{
if (g_stream_list[stream_id].stream_id == CM_INVALID_STREAM_ID) {
return CM_FALSE;
} else {
return CM_TRUE;
}
}
uint32 elc_stream_get_current_node()
{
return md_get_cur_node();
}
void elc_stream_lock_s(uint32 stream_id)
{
cm_latch_s(&g_stream_list[stream_id].latch, 0, CM_FALSE, NULL);
}
void elc_stream_lock_x(uint32 stream_id)
{
cm_latch_x(&g_stream_list[stream_id].latch, 0, NULL);
}
void elc_stream_unlock(uint32 stream_id)
{
cm_unlatch(&g_stream_list[stream_id].latch, NULL);
}
status_t elc_register_notify(usr_cb_status_notify_t cb_func)
{
g_cb_status_notify = cb_func;
return CM_SUCCESS;
}
status_t elc_register_election_notify(usr_cb_election_notify_t cb_func)
{
g_cb_election_notify = cb_func;
return CM_SUCCESS;
}
status_t elc_stream_change_leader_notify(uint32 stream_id, uint32 leader_id)
{
dcf_role_t role = g_stream_list[stream_id].node_role;
uint32 node_id = md_get_cur_node();
LOG_DEBUG_INF("[ELC]elc stream change leader, stream_id=%u local_node_id=%u new_leader=%u ",
stream_id, node_id, leader_id);
add_notify_item(stream_id, node_id, leader_id, role, role);
return CM_SUCCESS;
}
status_t elc_stream_refresh_hb_time(uint32 stream_id, uint64 leader_term, int32 leader_work_mode,
uint32 leader_id)
{
uint32 curr_node_id = elc_stream_get_current_node();
uint64 current_term = elc_stream_get_current_term(stream_id);
int32 work_mode = (int32)elc_stream_get_work_mode(stream_id);
if (leader_term < current_term && leader_work_mode == work_mode) {
LOG_DEBUG_INF("[ELC]leader's term less than current node, "
"leader_id=%u, leader_term=%llu, current_id=%u, current_term=%llu",
leader_id, leader_term, curr_node_id, current_term);
return CM_SUCCESS;
}
if (elc_stream_get_votefor(stream_id) != leader_id) {
LOG_RUN_INF("[ELC]receive new leader's heartbeat, stream_id=%u, node_id=%u, leader_id=%u",
stream_id, curr_node_id, leader_id);
dcf_role_t role = elc_stream_get_role(stream_id);
if (role == DCF_ROLE_CANDIDATE || role == DCF_ROLE_PRE_CANDIDATE || role == DCF_ROLE_LEADER) {
CM_RETURN_IFERR(elc_stream_set_role(stream_id, DCF_ROLE_FOLLOWER));
}
CM_RETURN_IFERR(elc_stream_set_votefor(stream_id, leader_id));
LOG_DEBUG_INF("[ELC]Set votefor when refresh hb time, stream_id=%u, votefor=%u", stream_id, leader_id);
CM_RETURN_IFERR(elc_stream_set_term(stream_id, leader_term));
} else if (leader_term > current_term) {
CM_RETURN_IFERR(elc_stream_set_term(stream_id, leader_term));
}
if (g_stream_list[stream_id].votefor_id != g_stream_list[stream_id].old_leader_id) {
CM_RETURN_IFERR(elc_stream_change_leader_notify(stream_id, g_stream_list[stream_id].votefor_id));
g_stream_list[stream_id].old_leader_id = g_stream_list[stream_id].votefor_id;
}
LOG_DEBUG_INF("[ELC]refresh heartbeat, leader_term=%llu, leader_id=%u, current_term=%llu, current_id=%u",
leader_term, leader_id, current_term, curr_node_id);
CM_RETURN_IFERR(elc_stream_set_timeout(stream_id, cm_clock_now()));
return CM_SUCCESS;
}
status_t elc_stream_refresh_hb_ack_time(uint32 stream_id, uint64 leader_term, uint32 node_id)
{
uint32 curr_node_id = elc_stream_get_current_node();
uint64 current_term = elc_stream_get_current_term(stream_id);
if (leader_term != current_term) {
LOG_DEBUG_INF("[ELC]invalid term, "
"node_id=%u, leader_term=%llu, current_node_id=%u, current_term=%llu",
node_id, leader_term, curr_node_id, current_term);
return CM_SUCCESS;
}
CM_RETURN_IFERR(elc_stream_set_hb_ack_time(stream_id, node_id, cm_clock_now()));
return CM_SUCCESS;
}
status_t elc_stream_set_work_mode(uint32 stream_id, dcf_work_mode_t work_mode, uint32 vote_num)
{
uint32 count;
uint32 node_id = md_get_cur_node();
if (work_mode == WM_MINORITY) {
CM_RETURN_IFERR(md_get_voter_num(stream_id, &count));
if (vote_num == 0 || vote_num > count) {
LOG_RUN_ERR("invalid vote_num:%u", vote_num);
return CM_ERROR;
}
g_stream_list[stream_id].vote_num = vote_num;
g_stream_list[stream_id].vote_node_work_mode[node_id] = g_stream_list[stream_id].work_mode;
g_stream_list[stream_id].work_mode = work_mode;
} else if (work_mode == WM_NORMAL) {
g_stream_list[stream_id].vote_num = 0;
g_stream_list[stream_id].vote_node_work_mode[node_id] = g_stream_list[stream_id].work_mode;
g_stream_list[stream_id].work_mode = work_mode;
} else {
LOG_RUN_ERR("invalid work_mode:%d", work_mode);
return CM_ERROR;
}
return CM_SUCCESS;
}
dcf_work_mode_t elc_stream_get_work_mode(uint32 stream_id)
{
return g_stream_list[stream_id].work_mode;
}
status_t elc_stream_set_vote_node_work_mode(uint32 stream_id, uint32 node_id, dcf_work_mode_t work_mode)
{
g_stream_list[stream_id].vote_node_work_mode[node_id] = work_mode;
return CM_SUCCESS;
}
dcf_work_mode_t elc_stream_get_vote_node_work_mode(uint32 stream_id, uint32 node_id)
{
return g_stream_list[stream_id].vote_node_work_mode[node_id];
}
timespec_t elc_stream_get_last_md_rep_time(uint32 stream_id)
{
return g_stream_list[stream_id].last_md_rep_time;
}
status_t elc_get_voting_weight(uint32 stream_id, uint32 node_id, uint32 *voting_weight)
{
if (elc_stream_get_work_mode(stream_id) == WM_MINORITY) {
*voting_weight = CM_ELC_NORS_WEIGHT;
} else {
CM_RETURN_IFERR(md_get_stream_node_weight(stream_id, node_id, voting_weight));
}
return CM_SUCCESS;
}
status_t elc_stream_set_last_md_rep_time(uint32 stream_id, timespec_t time)
{
g_stream_list[stream_id].last_md_rep_time = time;
return CM_SUCCESS;
}
bool32 elc_stream_is_force_vote(uint32 stream_id)
{
return g_stream_list[stream_id].force_vote;
}
void elc_stream_set_force_vote_flag(uint32 stream_id, bool32 is_force)
{
g_stream_list[stream_id].force_vote = is_force;
}
bool32 elc_stream_get_inter_promote_flag(uint32 stream_id)
{
return g_stream_list[stream_id].inter_promote_flag;
}
void elc_stream_set_inter_promote_flag(uint32 stream_id, bool32 flag)
{
g_stream_list[stream_id].inter_promote_flag = flag;
}
timespec_t elc_stream_get_leader_start_time(uint32 stream_id)
{
return g_stream_list[stream_id].leader_start_time;
}
void elc_stream_set_leader_start_time(uint32 stream_id, timespec_t time)
{
g_stream_list[stream_id].leader_start_time = time;
}
bool32 elc_stream_can_switch_now(uint32 stream_id)
{
timespec_t leader_start = g_stream_list[stream_id].leader_start_time;
timespec_t now = cm_clock_now();
uint64 interval = (now < leader_start) ? 0 : (uint64)(now - leader_start);
LOG_DEBUG_INF("[ELC]elc_switch_thresold now=%llu, leader_start=%llu, interval=%llu", now, leader_start, interval);
if (interval / MICROSECS_PER_SECOND >= elc_stream_get_elc_switch_thd_sec()) {
return CM_TRUE;
}
return CM_FALSE;
}
void elc_stream_set_leader_group(uint32 stream_id, uint32 leader_group)
{
g_stream_list[stream_id].leader_group = leader_group;
}
uint32 elc_stream_get_leader_group(uint32 stream_id)
{
return g_stream_list[stream_id].leader_group;
}
void elc_stream_set_my_group(uint32 stream_id, uint32 my_group)
{
g_stream_list[stream_id].my_group = my_group;
}
uint32 elc_stream_get_my_group(uint32 stream_id)
{
return g_stream_list[stream_id].my_group;
}
void elc_stream_set_priority(uint32 stream_id, uint64 priority)
{
g_stream_list[stream_id].priority = priority;
}
uint64 elc_stream_get_priority(uint32 stream_id)
{
return g_stream_list[stream_id].priority;
}
#ifdef __cplusplus
}
#endif