Raft_Shellmiao/node.py

import os
import json
import time
import random

from network_api import NetworkAPI
from logger import Log


class Node(object):
    def __init__(self, meta):
        self.role = 'follower'
        self.id = meta['id']
        self.addr = meta['addr']
        self.peers = meta['peers']
        self.path = "data/raft_front_end/public/log/"
        if not os.path.exists(self.path):
            os.makedirs(self.path)
        # 选举状态
        self.current_term = 0
        self.voted_for = None
        # 初始化选举情况
        self.load()
        # 初始化Logger
        log_name = self.path + '_' + self.id + "_log.json"
        self.log = Log(log_name)

        # volatile state
        # rule 1, 2
        self.commit_index = 0
        self.last_applied = 0

        # volatile state on leaders
        # rule 1, 2
        self.next_index = {_id: self.log.last_log_index + 1 for _id in self.peers}
        self.match_index = {_id: -1 for _id in self.peers}

        self.leader_id = None
        # 其他所有节点对于此节点的投票数
        self.vote_ids = {_id: 0 for _id in self.peers}

        self.client_addr = None

        self.wait_ms = (10, 20)
        self.next_leader_election_time = time.time() + random.randint(*self.wait_ms)
        self.next_heartbeat_time = 0

        # rpc
        self.network_api = NetworkAPI(self.addr, timeout=2)

    def load(self):
        """
        从文件中读取目前选举情况，如果文件不存在，则初始化
        :return: null
        """
        filename = self.path + "_" + self.id + '_persistent.json'
        if os.path.exists(filename):
            with open(filename, 'r') as f:
                data = json.load(f)

            self.current_term = data['current_term']
            self.voted_for = data['voted_for']
        else:
            self.save()

    def save(self):
        """
        将选举情况保存到json文件中
        :return: null
        """
        data = {'current_term': self.current_term,
                'voted_for': self.voted_for,
                }
        filename = self.path + "_" + self.id + '_persistent.json'
        with open(filename, 'w') as f:
            json.dump(data, f, indent=4)

    def redirect(self, data, addr):
        """
        按照data的类别以及此时该node的类别将address重定向到应有路径
        :param data:
        :param addr:
        :return:
        """
        if data is None:
            return None

        if data['type'] == 'client_append_entries':
            if self.role != 'leader':
                if self.leader_id:
                    # print('Hey,leader! Here is a message!')
                    self.network_api.send(data, self.peers[self.leader_id])
                return None
            else:
                return data

        if data['dst_id'] != self.id:
            self.network_api.send(data, self.peers[data['dst_id']])
            return None
        else:
            return data

    def append_entries(self, data):
        response = {'type': 'append_entries_response',
                    'src_id': self.id,
                    'dst_id': data['src_id'],
                    'term': self.current_term,
                    'success': False
                    }
        if data['term'] < self.current_term:
            response['success'] = False
            self.network_api.send(response, self.peers[data['src_id']])
            return

        self.leader_id = data['leader_id']

        # heartbeat
        if not data['entries']:
            return

        prev_log_index = data['prev_log_index']
        prev_log_term = data['prev_log_term']

        tmp_prev_log_term = self.log.get_log_term(prev_log_index)
        if tmp_prev_log_term != prev_log_term:
            response['success'] = False
            self.network_api.send(response, self.peers[data['src_id']])
            self.log.delete_entries(prev_log_index)
        else:
            response['success'] = True
            self.network_api.send(response, self.peers[data['src_id']])
            self.log.append_entries(prev_log_index, data['entries'])
            leader_commit = data['leader_commit']
            if leader_commit > self.commit_index:
                commit_index = min(leader_commit, self.log.last_log_index)
                self.commit_index = commit_index
        return

    def request_vote(self, data):
        response = {'type': 'request_vote_response',
                    'src_id': self.id,
                    'dst_id': data['src_id'],
                    'term': self.current_term,
                    'vote_granted': False
                    }
        if data['term'] < self.current_term:
            response['vote_granted'] = False
            self.network_api.send(response, self.peers[data['src_id']])
            return

        candidate_id = data['candidate_id']
        last_log_index = data['last_log_index']
        last_log_term = data['last_log_term']

        if self.voted_for is None or self.voted_for == candidate_id:
            if last_log_index >= self.log.last_log_index and last_log_term >= self.log.last_log_term:
                self.voted_for = data['src_id']
                self.save()
                response['vote_granted'] = True
                self.network_api.send(response, self.peers[data['src_id']])
            else:
                self.voted_for = None
                self.save()
                response['vote_granted'] = False
                self.network_api.send(response, self.peers[data['src_id']])
        else:
            response['vote_granted'] = False
            self.network_api.send(response, self.peers[data['src_id']])
        return

    def all_do(self, data):
        t = time.time()
        if self.commit_index > self.last_applied:
            self.last_applied = self.commit_index
        if data is None:
            return

        if data['type'] == 'client_append_entries':
            return

        if data['term'] > self.current_term:
            self.next_leader_election_time = t + random.randint(*self.wait_ms)
            self.role = 'follower'
            self.current_term = data['term']
            self.voted_for = None
            self.save()
        return

    def follower_do(self, data):
        t = time.time()
        if data is not None:
            if data['type'] == 'append_entries':
                if data['term'] == self.current_term:
                    self.next_leader_election_time = t + random.randint(*self.wait_ms)
                self.append_entries(data)
            elif data['type'] == 'request_vote':
                self.request_vote(data)
        if t > self.next_leader_election_time:
            self.next_leader_election_time = t + random.randint(*self.wait_ms)
            self.role = 'candidate'
            self.current_term += 1
            self.voted_for = self.id
            self.save()
            self.vote_ids = {_id: 0 for _id in self.peers}
        return

    def candidate_do(self, data):
        t = time.time()
        for dst_id in self.peers:
            if self.vote_ids[dst_id] == 0:
                request = {
                    'type': 'request_vote',
                    'src_id': self.id,
                    'dst_id': dst_id,
                    'term': self.current_term,
                    'candidate_id': self.id,
                    'last_log_index': self.log.last_log_index,
                    'last_log_term': self.log.last_log_term
                }
                self.network_api.send(request, self.peers[dst_id])
        if data is not None and data['term'] == self.current_term:
            if data['type'] == 'request_vote_response':
                self.vote_ids[data['src_id']] = data['vote_granted']
                vote_count = sum(list(self.vote_ids.values()))
                if vote_count >= len(self.peers) // 2:
                    self.role = 'leader'
                    self.voted_for = None
                    self.save()
                    self.next_heartbeat_time = 0
                    self.next_index = {_id: self.log.last_log_index + 1 for _id in self.peers}
                    self.match_index = {_id: 0 for _id in self.peers}
                    return
            elif data['type'] == 'append_entries':
                self.next_leader_election_time = t + random.randint(*self.wait_ms)
                self.role = 'follower'
                self.voted_for = None
                self.save()
                return
        if t > self.next_leader_election_time:
            self.next_leader_election_time = t + random.randint(*self.wait_ms)
            self.role = 'candidate'
            self.current_term += 1
            self.voted_for = self.id
            self.save()
            self.vote_ids = {_id: 0 for _id in self.peers}
            return

    def leader_do(self, data):
        t = time.time()
        # print('Here is leader!')
        if t > self.next_heartbeat_time:
            self.next_heartbeat_time = t + random.randint(0, 5)
            for dst_id in self.peers:
                request = {'type': 'append_entries',
                           'src_id': self.id,
                           'dst_id': dst_id,
                           'term': self.current_term,
                           'leader_id': self.id,
                           'prev_log_index': self.next_index[dst_id] - 1,
                           'prev_log_term': self.log.get_log_term(self.next_index[dst_id] - 1),
                           'entries': self.log.get_entries(self.next_index[dst_id]),
                           'leader_commit': self.commit_index
                           }
                self.network_api.send(request, self.peers[dst_id])
        if data is not None and data['type'] == 'client_append_entries':
            # print('Leader:Here is a message!')
            data['term'] = self.current_term
            self.log.append_entries(self.log.last_log_index, [data])
            return
        if data is not None and data['term'] == self.current_term:
            if data['type'] == 'append_entries_response':
                if not data['success']:
                    self.next_index[data['src_id']] -= 1
                else:
                    self.match_index[data['src_id']] = self.next_index[data['src_id']]
                    self.next_index[data['src_id']] = self.log.last_log_index + 1
        flag = False
        N = self.commit_index + 1
        count = 0
        for _id in self.match_index:
            if self.match_index[_id] >= N:
                count += 1
            if count >= len(self.peers) // 2:
                self.commit_index = N
                if self.client_addr:
                    response = {'index': self.commit_index}
                    self.network_api.send(response, self.client_addr)
                flag = True
                break

    def run(self):
        while True:
            # try:
            try:
                data, addr = self.network_api.receive()
            except Exception as e:
                data, addr = None, None
            data = self.redirect(data, addr)
            # print(data)
            self.all_do(data)

            if self.role == 'follower':
                self.follower_do(data)

            if self.role == 'candidate':
                self.candidate_do(data)

            if self.role == 'leader':
                self.leader_do(data)

            # except Exception as e:
            #     print(e)